[SPARK-59620][SQL] Late-materialization storage-filter pushdown in the vectorized Parquet reader by peter-toth · Pull Request #58895 · apache/spark

peter-toth · 2026-09-17T16:09:00Z

What changes were proposed in this pull request?

This adds a late-materialization read path to the vectorized Parquet reader, so the optimizer can push a runtime filter into the scan and have it prune value-column IO instead of running as a post-scan FilterExec. The filter it pushes today is BloomFilterMightContain, the runtime bloom InjectRuntimeFilter builds for a join.

How it works

Three phases per row group, all driven by one ParquetFileReader whose requested schema is switched between them:

Phase 0 narrows the row group to the rows the pushed data filters allow, through the column index. Metadata only, no page is read.
Phase 1 reads the key columns of those rows and evaluates the pushed conjunct per row, keeping the surviving key values.
Phase 2 reads the remaining columns, restricted to the pages those surviving rows fall in. A row group with no survivor at all is skipped without reading one value page of it.

Emit then splices each batch: the key columns come from what phase 1 kept, every other column from what phase 2 read.

Six things make the reader stop short, and all of them cost a saving rather than change an answer, because the conjunct is still in the post-scan Filter:

the surviving rows of a row group scatter into more row ranges than the memory budget allows, so that row group is read with no filter applied;
the key values buffered for splicing pass the same budget, so they are released and phase 2 reads every projected column of the surviving rows instead, at the cost of reading the key columns twice;
evaluating the conjunct raises an error, which it can only do on a row the conjuncts ahead of it would have rejected, so that row group is read with no filter applied and the Filter decides in its own order;
the file has no Parquet page index, so phase 2 cannot narrow to part of a row group and the filter is left with the row groups it empties;
parquet.filter.columnindex.enabled is false, which says the file's page index is not to be trusted, so the filter is not applied at all;
the vectorized reader turns out to be unavailable at execution time, so the read is plain from the start.

Design decisions

This is the first step of a longer effort, and it is deliberately narrow: the v1 Parquet file source, and the bloom filter InjectRuntimeFilter already produces. The steps after it are a DSv2 equivalent, so a connector such as Iceberg can prune the same way, and per-partition filters for a storage-partitioned join. What all three share is the goal here: a scan that is handed a runtime filter and uses it to decide what not to read, instead of reading everything and dropping rows above itself.

A pushed conjunct stays in the post-scan Filter as well. The scan is offered it, not obliged to honor it, exactly as a pushed data filter works today: FileSourceStrategy removes only partition filters from the plan, because partition pruning is exact, and leaves the data filters it hands to Parquet in place, because statistics and column-index pruning are not. Late materialization is the second kind, and DSv2 draws the same line from the other side: a source returns the filters it cannot guarantee, and PushDownUtils keeps those post-scan.

What that buys is the list above: a reader free to stop short whenever honoring the filter would cost more than it saves, or would raise an error the plan would not, with no correctness argument attached to any of it. It also means the bloom is never built for nothing, since the Filter uses it whatever the scan managed to do. Both sides read the same bytes: the subquery that builds it is executed once, deduplicated by AQE or by subquery reuse, both on by default.

What it costs is evaluating the conjunct a second time for the rows the scan emits. Where the filter was applied those are the surviving rows, so the cost is proportional to what it kept rather than to what it discarded, and it is a hash and a few bit probes fused into the generated code. Where the reader stopped short they are all the rows the pushed data filters left, which is what the plan would have evaluated anyway.

So turning the conf on can cost a slower read of a row group the reader cannot prune, never a wrong answer. Nor a failed query: an error from evaluating the conjunct gives that row group up, and the errors this conjunct can raise are Spark's own, since InjectRuntimeFilter will not build a bloom over a UDF or any other expression it calls unsimple. An error carrying no error class is a defect in the reader, and that still fails the query.

Why does phase 2 splice buffered key values instead of re-reading the key columns? Re-reading is the simpler design and needs none of the splicing machinery, but it pays a second read of the key columns for every row group. Nothing absorbs that read on object storage: parquet caches footers, not pages, and S3A's default input stream caches nothing either, so it is a new GET rather than a page-cache hit. The stream types that do cache are opt-in (fs.s3a.prefetch.enabled defaults to false, fs.s3a.input.stream.type to classic).

Measured on a synthetic table of one long key and twenty long payload columns written in 256-row pages: re-reading transfers 4.7% more bytes than splicing, and 49% more when the projection is the key and one payload column. The comparison that decides it is against the feature being off rather than against splicing. Phase 1 has already read the key column once, so re-reading makes the feature read more than not having it at all, while splicing's worst case is exactly a plain read.

Why is an all-keys projection not pushed? A scan whose projected data columns are all key columns of the filter reads exactly the same columns for the same rows as a plain scan, because the reader has to read a key column to evaluate the filter on it, and phase 2 is then left with nothing to prune. All it can add is cost: phase 1 evaluates the predicate per row through BasePredicate.eval on a batch row instead of having it fused into the scan's generated code, the way a post-scan FilterExec does, plus one value copy per survivor. So the planner declines that shape. Making phase 1 evaluate over the key vector is a follow-up, and one of the reasons the conf is off by default.

Why is the survivor buffer bounded per row group rather than evaluated in chunks? Phase 1 evaluates a whole row group before the first batch of that row group is emitted, so the survivor vectors are the one thing a task holds that scales with row-group size rather than with the batch capacity. Phase 1 counts the bytes it has buffered and, past spark.sql.parquet.storageFilterPushdown.maxSplicedRowGroupBytes (internal, 64MB), releases them and gives that row group up: phase 2 then reads every projected column of the surviving rows, so nothing is buffered and the cost is one extra read of the key columns. Evaluating one capacity-sized chunk at a time would bound the memory too, but it turns phase 2 into one read call per chunk, refetches any page that straddles a chunk boundary, and gives up the coalescing and parallelism that a single vectored read gets. What it would shorten is how long the first batch of a row group waits, which is one row group's key decode. The IO is unchanged, since parquet reads a row group's whole requested chunks before it hands back a page store.

Planning

New SQL conf spark.sql.parquet.storageFilterPushdown.enabled, default false, read by ParquetFileFormat rather than by the planner so a Parquet-named conf never decides for another format. It is a planning-time decision: with it off nothing is attached to a scan and the bloom stays where it is today.
FileSourceStrategy.storageFiltersFor picks the conjuncts of afterScanFilters to offer, and they go into a new storageFilters slot on FileSourceScanExec while staying in the Filter. Two conditions are per scan, the format's own supportsStorageFilterPushdown and supportBatch for the schema the reader will see. The rest are per conjunct: deterministic, referencing at least one column and only projected data columns, and accepted by the format. One more is on the set that survives: at least one projected data column must not be a key column, or there is nothing for the reader to prune.
FileFormat.supportsStorageFilter(expr), defaulting to false, is how the format accepts a conjunct, the same shape as supportBatch and vectorTypes. The planner keeps what it can see in the plan and asks the format about everything to do with its own reader, so the expression shapes and column types a reader supports stay in that reader's package and FileSourceStrategy names no format. ParquetFileFormat answers true for a top-level BloomFilterMightContain over a key type its reader can copy, and false for a subclass of itself, which may customize reading in a way a storage-filter scan would bypass.
FileSourceScanLike gains the slot and five SQL metrics, described below, and FileSourceScanExec.preparedStorageFilters materializes scalar subqueries and binds the references before the reader is built. The conf is deliberately not rechecked at execution time: a conf flipped in between makes the reader read plainly, not fail.

FileFormat API

A new buildReaderWithStorageFilters overload takes the filters and the metric map, and returns an Option. A format that does not apply them answers None, which is the default, and FileSourceScanExec.inputRDD then builds an ordinary reader. Declining is safe for the same reason a reader may stop short: the plan still holds the conjunct. The Option is also what keeps the two builders from being able to call each other, which delegating from the default body would have invited.
FileSourceScanExec.inputRDD routes through the new entry point only when there is something to push, so a ParquetFileFormat subclass that customizes buildReaderWithPartitionValues keeps working unchanged.

Parquet reader

A new ParquetStorageFilter holds the bound expression, the key-column indices into the requested schema, and the metrics. rewriteForMissingKeys and evalAllMissing handle schema evolution: a key column that is in the requested schema but not in the file is substituted with the value the reader will materialize for it, its existence DEFAULT or null, because filtering on anything else would filter on a value the scan never returns.
SpecificParquetRecordReaderBase exposes the underlying ParquetFileReader, which is what lets the three phases switch the requested schema and call readFilteredRowGroup(blockIdx, rowRanges).
ParquetReadState now coalesces the row indexes it walks into ranges lazily, holding the current one as two longs. It consumes its ranges once, in order, so the list it used to build bought nothing and cost one per column reader of the row group, with one entry per surviving row in each. Every read that uses a column index is lighter for it, and it is what keeps the budget below out of the way in all but pathological cases.
One owner weighs that budget: phase 1 checks it after every surviving row, over the buffered key bytes and the row ranges together.
parquet.filter.columnindex.enabled set to false turns the whole path off. That conf is the escape hatch for a file whose page index is wrong, and phase 2 reads part of a row group through the offset index whatever it says, so honoring it in phase 0 alone would leave a wrong index able to pair a row's key with another row's values.
The page stores the phases read from are closed by this reader, since readFilteredRowGroup hands out a store the file reader does not track, unlike readNextRowGroup.
What planning already guarantees is asserted rather than handled: a key ordinal out of range, a non-primitive key column, no key pages for a block the pushed filter kept. Those are internal errors, not conditions of a file.

Metrics: five, all created only when the scan has storage filters, and all scoped to what the storage filter added on top of a no-storage-filter read of the same projection.

storageFilterRowGroupsSkipped, "row groups skipped by storage filter", counts row groups whose data columns were never read, though phase 1 did read their key columns.
storageFilterRowsExcludedByRowGroup, "rows excluded by storage filter (whole row group)".
storageFilterRowsExcludedWithinRowGroup, "rows excluded by storage filter (within row group)". The suffix says where the row was excluded rather than by which mechanism, because a row sharing a page with a survivor is read and dropped during decode.
storageFilterBytesAvoidedByRowGroup, "bytes avoided by storage filter (whole row group)".
storageFilterBytesAvoidedByPageFiltering, "bytes avoided by storage filter (page filtering)".

The byte counters must not cost IO to report, and they do not: the whole-block case answers from the footer's ColumnChunkMetaData.getTotalSize(), and a narrower range can only exist where something has already built the block's offset index, so the walk is metadata arithmetic either way. It counts the dictionary page too, since parquet reads that whenever it reads any data page of a chunk.

Why are the changes needed?

A runtime bloom filter from join runtime filtering is applied as a post-scan FilterExec today. The scan still reads every value page of every row group, even where the bloom drops almost every row immediately. On a selective join over a wide table that read is the dominant cost.

Late materialization turns that around. The scan reads the bloom's key column first, decides which rows survive, and never reads the value pages no surviving row touches. A row group where nothing survives costs one key-column read and no value IO at all.

Does this PR introduce any user-facing change?

No change by default, since the conf is off and nothing is attached to a scan in that case.

With the conf on:

Queries return the same rows. The late-materialization path is exact, and a filter that fails any eligibility check keeps its existing post-scan FilterExec.
FileSourceScanExec reports the five metrics above in the SQL UI, and its description gains a StorageFilters entry.
A row group the reader cannot prune is read the way a plain scan reads it, plus one more read of its key columns, since the phase that evaluated the filter has already read them. The query is correct and pays that. On a file written with no Parquet page index, every row group that keeps a row pays it, since narrowing to part of one needs that index: parquet-mr has written the page index since 1.11, but other writers do not, and pyarrow's write_table defaults to write_page_index=False.
Two new WARN lines, at most one of each per file: one when a row group's filter is given up because the file has no offset index, one when evaluating the filter raised an error.
A consumer that illegally retains a ColumnarBatch across next() sees a sharper edge than before. The batch object is the same one throughout, as resultBatch() documents, but its key slots are rewritten per batch and the vector a previous batch used is freed at the next nextBatch(), so with off-heap vectors a retained reference points at released memory rather than stale values. The contract already forbids retaining a batch.
Phase 1 buffers a whole row group's surviving key values before it produces that row group's first batch, so a task holds up to one extra copy of the key columns for one row group. spark.sql.parquet.storageFilterPushdown.maxSplicedRowGroupBytes bounds the values and their per-row overhead, not the backing arrays a column vector may grow beyond that.

How was this patch tested?

New ParquetStorageFilterSuite, 99 tests. Four blind review rounds ran over the change before it was opened, each by an agent with no knowledge of the earlier ones, and every finding from those and from the reviews on the PR is fixed.

Reader-level tests over hand-built filters: whole row group rejected, nothing rejected, mixed, multi-batch emit, key-only projection, a survivor count that is an exact multiple of the batch capacity, and a projection whose key column is not in the leading slot.
One case per ValueCopier branch across 17 types, each in both of the encodings parquet will actually write for it, and each comparing the splicing path against the plain reader over the same file, so values are checked and not only row counts. The encoding is asserted rather than assumed, because asking for a dictionary does not mean getting one: parquet falls back to PLAIN once the dictionary is no smaller than the values it replaces, and it has no dictionary writer at all for BOOLEAN or for a byte-array DECIMAL. isSupportedKeyType is asserted to cover exactly the copier's types.
Nullable keys, two key columns, partition columns alongside spliced keys, off-heap vectors, the row-at-a-time path, _metadata.row_index, and a complex non-key column.
Schema evolution: rewriteForMissingKeys and evalAllMissing as units, plus end-to-end reads where a key column is missing from the older file with a DEFAULT, with a DEFAULT that fails the filter, and with no DEFAULT.
Page-level ranges, which need a multi-page row group: phase 1 stays aligned with the row indexes, and the byte metrics still credit a saving. Asserting they are non-negative would prove nothing, since SQLMetric drops a negative.
Metric arithmetic: emitted plus excluded accounts for every row of the file, and an all-key projection reports zero avoided bytes.
Planner gates: the bloom stays in the post-scan Filter when the vectorized reader is unavailable, when the conf is off, when the conjunct is non-deterministic, and when every projected data column is a key column. supportsStorageFilterPushdown is asserted with the conf off and on and for a ParquetFileFormat subclass, and supportsStorageFilter for a plain bloom, a non-bloom conjunct, a bloom over an unsupported key type, a cast key, a non-deterministic value side, and the trait's default.
The ANSI case end to end, with the non-numeric row inside the same row group as matching ones so no statistic prunes it away: the cast key is pushed, the evaluation throws, the row group is given up, and the query returns its rows. Ablating the fail-open makes the same test fail with CAST_INVALID_INPUT, which is how the scenario was confirmed rather than assumed.
Extraction preserves results with AQE on and off, and the AQE arm asserts the filter is still on the scan of the final adaptive plan: equal results prove nothing there, since the post-scan Filter guarantees them. Canonicalization keeps a storage-filter scan distinct from a plain one, so exchange and subquery reuse cannot cross them. The feature still engages with ignoreCorruptFiles on, and its metrics are what say so.
Every way the reader stops short. A vectorized-reader conf flipped after planning returns every row of the file. A row group whose survivors scatter into more ranges than the cap allows returns that row group whole, whether the cap is crossed inside phase 1's survivor loop or by the ranges alone. And a file with no offset index, built with the low-level ParquetFileWriter overloads that take no row count: its emptied row group is still skipped whole while the other is read whole.
The survivor cap, for a mixed and an all-keys projection: past it the rows are unchanged and the read transfers strictly more bytes, which is what shows the row group really did give splicing up rather than the predicate having been dropped. And one file whose first row group splices while a later one passes the cap, since the batch is one object and its key slots have to go back to the vectors phase 2 reads into.
Whole-stage codegen off, where the planner's gate is weaker than the runtime's, so the bloom is extracted and the spliced batch is served one row at a time.
Column-index filtering off, where the reader must not apply the filter at all. The metrics say so, and the arm with it on is asserted to report something, so the test cannot pass by the feature being off everywhere.
A limit under whole-stage codegen, which is the only shape that closes the spliced batch from outside while the reader is still open. The test asserts the generated source contains that close, so it cannot pass for the wrong reason.
The byte metrics cost no extra IO, measured rather than argued. The same read runs twice over a filesystem that counts every byte handed back, once with all five metrics wired and once with none, and the counts match. Both range shapes are covered, the offset-index one and the footer one.

Knowingly untested: the ParquetFileFormat-subclass bypass, which needs a third-party subclass.

Measured on a real workload's data, not on TPCDS. One file of ~19.5k rows, read as a plain parquet directory, joined against a dimension side sampled out of its own key values so the selectivity is exact. The shape is what makes it interesting: the join key averages 43 bytes per row while the single projected value column averages ~53 kB per row, and the file was written with a 1 MB page target, which came out at ~4 rows per value page and ~1,500 per row group. So phase 1 reads 833 kB to decide about 991 MB. The query is written to the noop sink, so every surviving value is materialized and nothing is written. Both arms are asserted to return the same rows and the same total value bytes, and the bytes are counted inside a wrapping filesystem as the reader takes them. Local SSD, timings vary 10-30% run to run while the byte counts do not move.

Key selectivity	Matching rows	Bytes read, feature off	Bytes read, feature on	Bytes avoided	Wall clock off	Wall clock on	Speedup	Row groups skipped whole
0.1%	20	991 MB	6 MB	99.4%	3592 ms	286 ms	12.6x	4 of 13
0.5%	98	991 MB	50 MB	95.0%	2089 ms	263 ms	7.9x	0
1%	195	991 MB	68 MB	93.1%	1983 ms	416 ms	4.8x	0
5%	976	991 MB	250 MB	74.7%	2412 ms	1216 ms	2.0x	0

Two things this table does not say. It is a fine-page layout, and the page size is what decides the saving, since one surviving row pulls in its whole page: the same data written with pages of a few hundred rows saves a fraction of this at 0.1% and nothing at 5%, which is one of the reasons the conf is off by default and a gating heuristic is a follow-up. And the wall-clock column is this shape's, not a general claim: rows of tens of kilobytes mean a skipped page skips megabytes of decode as well as the read, which is not true of a narrow table.

The benchmark suite is not part of this PR, since it needs data that cannot ship with it.

Possible follow-ups:

Evaluating the filter over the key column vector rather than per row. The predicate is already compiled, but phase 1 calls it once per row through a row view of the batch, where a post-scan Filter has it inlined in the loop that produced the row. That is the per-row overhead an all-keys projection pays for nothing, and part of why the conf is off by default.
A gating heuristic, so the conf can eventually default to on: the all-keys condition is its trivial first slice, and the rest is projection width against expected selectivity.
A vector-to-vector appendBytes on WritableColumnVector, which would remove the per-row byte[] the string copier allocates. It touches the core column-vector classes, so it belongs in its own change.
Degrading inside parquet on a file with no page index: readFilteredRowGroup could read whole chunks while keeping the row ranges for getRowIndexes(), which would let the filter apply where this PR gives it up.
Offering the conjuncts that bring in no new column, so a WHERE k > 100 next to a bloom on k is filtered in the reader too, at no extra read. The general form, a prefix of the plan's condition, would also remove the evaluation-order question entirely, but it makes every column the prefix touches a key column, so it needs a cost model rather than a rule.
Broadcasting the bloom bytes once and sharing the deserialized filter per executor, the way InSubqueryExec does. The conjunct staying in the plan means the task binary now carries the filter twice, and each task deserializes both.
A PageReadStore.getRowRanges() in parquet. The store is built from a RowRanges and flattens it into the row-index iterator this reader then coalesces back, so exposing it would delete the coalescing outright.
Letting a format declare its own storage-filter metrics. The five listed above are Parquet's vocabulary in a format-agnostic class, which is the right shape to revisit when a second format implements this.
Coarsening the surviving row ranges rather than giving the filter up when their list would pass the budget. Closing a gap shorter than the smallest page row count of the columns phase 2 reads costs no extra bytes, since no page fits wholly inside such a gap, so the filter would keep applying where this PR drops it.
Accounting the survivor buffer through a MemoryConsumer whose spill() gives splicing up, so the unified memory manager sees it.
Letting the post-scan Filter skip the rows the scan already checked, by having the scan return a column that says so. That removes the second evaluation this design pays for, at the price of a synthetic column threaded through the batch and a condition rewrite, and it is the shape the two steps after this one want.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code with Claude Opus 4.7 and Claude Opus 5

Co-authored-by: Matt Butrovich mbutrovich@gmail.com

peter-toth · 2026-09-17T16:09:15Z

Opening this as a draft because it depends on #58120, which upgrades Parquet to 1.18.1. The feature needs two APIs that arrive with that upgrade, RowRanges.Builder and the public ParquetFileReader.getRowRanges(int).

This branch is stacked on #58120's head, so that PR's commit shows up in the diff here until it merges. Only the second commit is mine. I will rebase onto master and take this out of draft once #58120 is in.

…e vectorized Parquet reader ### What changes were proposed in this pull request? This adds a late-materialization read path to the vectorized Parquet reader, so the optimizer can push a runtime filter into the scan and have it prune value-column IO instead of running as a post-scan `FilterExec`. The filter it pushes today is `BloomFilterMightContain`, the runtime bloom `InjectRuntimeFilter` builds for a join. The reader reads the filter's key columns first, evaluates the predicate per row, and then reads the remaining columns restricted to the surviving row ranges. Each output batch is spliced together from the key vectors it kept and the value vectors it read for those rows. **Planning.** - New SQL conf `spark.sql.parquet.storageFilterPushdown.enabled`, default `false`, session-bound. It is a planning-time decision only. With it off, no storage filter is attached to a scan in the first place and the bloom stays where it is today. - `FileSourceStrategy.extractStorageFilters` lifts eligible top-level `BloomFilterMightContain` conjuncts out of `afterScanFilters` into a new `storageFilters` slot on `FileSourceScanExec`. Eligibility is checked in full at planning time, because extraction removes the conjunct from the post-scan `Filter` and nothing else would apply it afterwards. A conjunct qualifies when the file format is exactly `ParquetFileFormat`, the vectorized reader is feasible for `partitionSchema ++ outputDataSchema`, the conjunct is deterministic, and every reference on the bloom's value side is a projected data column of a type the reader can copy. - `ParquetStorageFilter.isSupportedKeyType` is the single authority on key-column types. Both `extractStorageFilters` and `ParquetStorageFilter.create` consult it, and it lists exactly the types `VectorizedParquetRecordReader.copierFor` handles. It is deliberately narrower than `AtomicType`, since `VariantType`, `GeometryType` and `GeographyType` are atomic but have no primitive Parquet leaf for phase 1 to read into. - `FileSourceScanLike` gains `storageFilters: Seq[Expression]` and five SQL metrics, described below. The `StorageFilters` entry in the scan description is only emitted when the scan has storage filters, so explain output is unchanged for everyone else. - `FileSourceScanExec.preparedStorageFilters` materializes scalar subqueries and binds attributes to `BoundReference`s indexing `requiredSchema`. The conf is deliberately not rechecked at execution time. Once extraction has dropped a bloom from the post-scan `Filter`, the runtime has to honour that decision or fail, and it does the second. **`FileFormat` API.** - A new `buildReaderWithStorageFilters` overload takes `storageFilters: Seq[Expression]` and the metric map. Its default body requires `storageFilters` to be empty and otherwise delegates to `buildReaderWithPartitionValues`, so a format that does not implement the feature rejects a filter it cannot honour rather than dropping it. - `FileSourceScanExec.inputRDD` only routes through the new entry point when there is something to push, so a `ParquetFileFormat` subclass that customizes reading by overriding `buildReaderWithPartitionValues` keeps working unchanged. **Parquet reader.** - A new `ParquetStorageFilter` value object holds the bound expressions, the key-column indices into the requested schema, and the optional metrics. `rewriteForMissingKeys` and `evalAllMissing` handle schema evolution, where a key column is in the requested schema but absent from the physical file. The rewritten predicate is evaluated against the value the reader will actually materialize for that column, which is the column's existence `DEFAULT` when it has one and null otherwise. Evaluating against null instead would filter on a value the scan never returns, and `XxHash64` is `nullable = false` so a null input hashes to the seed rather than producing null. - `SpecificParquetRecordReaderBase` exposes the underlying `ParquetFileReader`, the input file and the footer, so the late-materialization driver can switch the requested schema per phase and call `readFilteredRowGroup(blockIdx, rowRanges)`. - `VectorizedParquetRecordReader` gains a three-phase per-row-group loop driven by that one reader. Phase 0 computes `pushedFilterRanges` from the pushed data filter through the column index, which is metadata only. Phase 1 switches to the key-only schema, reads the key columns under those ranges, evaluates the storage filter per row, and accumulates the survivors into per-key-column queues of capacity-sized `WritableColumnVector`s. Phase 2 switches to the non-key schema and reads those columns under the surviving ranges. Emit splices the dequeued key vectors with the freshly read non-key vectors in the projection's own order. - A projection that is all key columns skips phase 2 entirely and reconstructs every batch from the key queues, which is the shape with the largest saving. - Phase 0 checks `parquet.filter.columnindex.enabled` itself, because `ParquetFileReader.getRowRanges` only asks whether a filter is pushed. That conf is the documented escape hatch for a file whose column index is wrong, and trusting a wrong column index here would drop rows for good. - `requireOffsetIndexesForPhase2` fails at reader init if any projected column of any row group has no offset index. Phase 2 reads a strict subset of a row group's rows, which parquet can only do through the offset index, and neither widening the read nor skipping the filter is correct. The check reads only footer fields, and it covers every projected column rather than the non-key ones alone, because parquet builds one column index store per row group and returns an empty one as soon as any path in it lacks an offset index. - Per-key-column value copying uses a `ValueCopier` chosen once at init, so the survivor loop has no per-value type dispatch. - `initBatch` threads a `skipDataSlots` set through `allocateColumns`, so the persistent output vectors for key slots are never allocated. Under splicing those slots come from the queues. - Preconditions that planning already guarantees throw instead of falling back, for the same reason the conf is not rechecked. That covers a key ordinal out of range, a non-primitive key column, missing key pages, and a vectorized-reader conf flipped between planning and execution. The two remaining silent fallbacks cannot change the result, namely a reader with no underlying `ParquetFileReader`, which only test mocks produce, and a file where every key column is missing, which is answered by evaluating the rewritten constant predicate. **Metrics.** Five, all created only when the scan has storage filters, and all scoped to what the storage filter added on top of a no-storage-filter read of the same projection. Each counter names its own quantity, so the three verbs are deliberate. A row group is skipped, meaning its data columns were never read while phase 1 did read its key columns. A row is excluded, meaning it never reached the output. A byte is avoided, meaning it was never transferred. - `storageFilterRowGroupsSkipped`, "row groups skipped by storage filter". - `storageFilterRowsExcludedByRowGroup`, "rows excluded by storage filter (whole row group)". - `storageFilterRowsExcludedWithinRowGroup`, "rows excluded by storage filter (within row group)". The suffix says where the row was excluded rather than by which mechanism, because a row sharing a page with a survivor is read and dropped during decode. - `storageFilterBytesAvoidedByRowGroup`, "bytes avoided by storage filter (whole row group)". - `storageFilterBytesAvoidedByPageFiltering`, "bytes avoided by storage filter (page filtering)". The byte counters must not cost IO to report, so `compressedBytesForRowRanges` answers from the footer's `ColumnChunkMetaData.getTotalSize()` whenever the row range covers the whole block, and walks the offset index only for a strict subset. That is complete rather than a mitigation. A range narrower than the block can only come from column-index filtering, which builds and memoizes the store as a side effect, and phase 2's own read builds it before the walk in the other case. The walk counts the dictionary page too, since parquet reads it whenever it reads any data page of a chunk. ### Why are the changes needed? A runtime bloom filter from join runtime filtering is applied as a post-scan `FilterExec` today. The scan still reads every value page of every row group, even where the bloom drops almost every row immediately. On a selective join over a wide table that read is the dominant cost. Late materialization turns that around. The scan reads the bloom's key column first, decides which rows survive, and never reads the value pages no surviving row touches. A row group where nothing survives costs one key-column read and no value IO at all. ### Does this PR introduce _any_ user-facing change? No change by default, since the conf is off and nothing is attached to a scan in that case. With the conf on: - Queries return the same rows. The late-materialization path is exact, and a filter that fails any eligibility check keeps its existing post-scan `FilterExec`. - `FileSourceScanExec` reports the five metrics above in the SQL UI, and its description gains a `StorageFilters` entry. - A scan reading a Parquet file written without offset indexes fails with an error naming the conf to turn off. Files written by parquet-mr 1.11 and later always have them. - A consumer that illegally retains a `ColumnarBatch` across `next()` sees a sharper edge than before. On the plain path the previous batch's key vectors are reused, and under splicing they are freed at the next `nextBatch()`, so with off-heap vectors the retained reference points at released memory rather than stale values. The contract already forbids retaining a batch. - Phase 1 buffers a whole row group's surviving key values before it produces that row group's first batch, so a task holds up to one extra copy of the key columns for one row group. The conf's documentation says so. ### How was this patch tested? New `ParquetStorageFilterSuite`, 89 tests. Four blind review rounds ran over the change, each by an agent with no knowledge of the earlier ones, and every finding is fixed in this commit. - Reader-level tests over hand-built filters: whole row group rejected, nothing rejected, mixed, multi-batch emit, key-only projection, a survivor count that is an exact multiple of the batch capacity, and a projection whose key column is not in the leading slot. - One case per `ValueCopier` branch across 17 types and both encodings, each comparing the splicing path against the plain reader over the same file, so values are checked and not only row counts. `isSupportedKeyType` is asserted to cover exactly the copier's types. - Nullable keys, two key columns, partition columns alongside spliced keys, off-heap vectors, the row-at-a-time path, `_metadata.row_index`, and a complex non-key column. - Schema evolution: `rewriteForMissingKeys` and `evalAllMissing` as units, plus end-to-end reads where a key column is missing from the older file with a `DEFAULT`, with a `DEFAULT` that fails the filter, and with no `DEFAULT`. - Page-level ranges, which need a multi-page row group: phase 1 stays aligned with the row indexes, and the byte metrics stay non-negative. - Metric arithmetic: emitted plus excluded accounts for every row of the file, and an all-key projection reports zero avoided bytes. - Planner gates: the bloom stays in the post-scan `Filter` when the vectorized reader is unavailable, when the conf is off, and when the conjunct is non-deterministic. A scan whose vectorized reader is disabled after planning fails loudly. Extraction preserves results with AQE on and off. Canonicalization keeps a storage-filter scan distinct from a plain one, so exchange and subquery reuse cannot cross them. - Whole-stage codegen off, where the planner's gate is weaker than the runtime's, so the bloom is extracted and the spliced batch is served one row at a time. - Column-index filtering off, the branch that decides where phase 0's ranges come from. The row accounting tells the two arms apart, since with the column index off every row of the block reaches phase 1. - A limit under whole-stage codegen, which is the only shape that closes the spliced batch from outside while the reader is still open. The test asserts the generated source contains that close, so it cannot pass for the wrong reason. - The byte metrics cost no extra IO, measured rather than argued. The same read runs twice over a filesystem that counts every byte handed back, once with all five metrics wired and once with none, and the counts match. Both range shapes are covered, the offset-index one and the footer one. Knowingly untested, both for the same reason. The offset-index check's throw needs a file with no offset index, which parquet-mr cannot write. The `ParquetFileFormat`-subclass bypass needs a third-party subclass. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code with Claude Opus 4.7 and Claude Opus 5 Co-authored-by: Matt Butrovich <mbutrovich@gmail.com>

peter-toth · 2026-09-22T12:05:44Z

#58120 has merged, so this is now rebased onto master and out of draft.

peter-toth · 2026-09-22T12:08:21Z

@cloud-fan, @dongjoon-hyun, @sunchao, can you please take a look at this PR?

dongjoon-hyun

Thanks for this -- the late-materialization design is clearly thought through, and the test suite is unusually thorough for a reader change of this size. I went through it in detail and left 15 inline comments. Summary of the ones I think matter most, plus one design question.

Two resource-lifecycle bugs

Dequeued key vectors leak if phase 2 throws. In nextBatchSplicing, the survivor vectors are pulled out of keyVectorQueues ~40 lines before pendingCloseKeyVectors = dequeued is assigned. Anything that throws in between (corrupt page, object-store read error) leaves them in neither the queues nor pendingCloseKeyVectors, so closeSplicingState() never frees them. Off-heap, that is a permanent native leak on each of the 4 task attempts. One-line fix: assign pendingCloseKeyVectors right after the dequeue loop.

keyScratchVectors are unreachable from close() if the allocation loop throws. ensureKeyScratchAllocated() assigns keyScratchBatch only after the loop completes, and close() frees the scratch vectors solely through keyScratchBatch. Closing keyScratchVectors element-wise, the way closeSplicingState() already does for the accumulators, removes this.

`requireOffsetIndexesForPhase2()` makes this an availability feature, not a performance one

Files without a page index -- pyarrow's write_table defaults to write_page_index=False, and parquet-mr < 1.11 has no offset index at all -- now fail the whole query with IllegalStateException, where before this PR the bloom was an ordinary post-scan FilterExec and the query simply worked. One legacy file in one partition is enough, the check fires even when the filter keeps every row and phase 2 would never touch the index, and the only remedy offered is a global conf flip. The reader already has precedent for degrading gracefully per file (the all-keys-missing path sets storageFilter = null); a per-file fallback to the eager path would keep this a pure optimization.

Unbounded, unaccounted survivor buffering

Phase 1 buffers every surviving key value of a whole row group before the first batch is emitted. At the default 128MB parquet.block.size that is commonly 1M+ rows, and blooms are deliberately false-positive-prone, so most survive. Those vectors are allocated outside any MemoryConsumer, so with off-heap enabled the executor is OOM-killed rather than raising SparkOutOfMemoryError. The conf doc's "one extra copy of the key columns for one row group" is accurate but reads as bounded; it is not, and for a dictionary-encoded key the accumulators hold expanded bytes. The same buffering also removes LIMIT short-circuiting -- the early-termination tests all use rowGroupSize = 256L, so they do not exercise it.

Design question: why buffer instead of re-read?

This is the main thing I would like to understand before the rest is worth polishing. The entire splicing apparatus exists only because phase 2 deliberately omits the key columns: nextBatchSplicing, ValueCopier/copierFor, the five accumulator/queue helpers, the double-close handling in close(), the skipDataSlots parameter, eight fields, and ParquetStorageFilter.isSupportedKeyType with the type gate it forces into FileSourceStrategy -- roughly 400 lines.

If phase 2 read the full requestedSchema under finalRanges, the key columns would arrive from the same PageReadStore, already aligned, and the existing eager nextBatch() would serve every row group unchanged. That also removes the buffering, the LIMIT stall, the per-row byte[] copy for string keys, and the load-bearing "keyColumnIndices must stay sorted" coupling that the code's own comment warns can silently swap key columns.

The honest cost is re-reading one narrow column's pages under a subset of the ranges phase 1 already read -- always cheaper than phase 1 itself, usually still in page cache, worst case 2x key decode when everything passes. The PR does not discuss this trade anywhere; if there is a reason it does not work, it would be worth a comment in the code, since it is the first thing a reader of nextBatchSplicing will wonder.

Smaller things

An API hazard worth closing while it is cheap: ParquetFileFormat.buildReaderWithPartitionValues now delegates to the overridable buildReaderWithStorageFilters, whose FileFormat default delegates back. A subclass that overrides the new method and calls buildReaderWithPartitionValues -- or even super.buildReaderWithPartitionValues -- recurses forever. Not reachable in-tree, but subclassing ParquetFileFormat is a common downstream pattern.

Also inline: resultBatch()'s "This object is reused" contract is no longer true under splicing; the two new protected fields on SpecificParquetRecordReaderBase are never read by anyone; copierFor duplicates RowToColumnConverter (and has already drifted -- the VarcharType/CharType branches are dead since both extend StringType); MessageType.getColumns() is rebuilt ~9x per row group for two metrics; .version("5.0.0") should probably be 4.4.0; and the planner's getClass != classOf[ParquetFileFormat] would read better as a FileFormat capability method, the way supportBatch right next to it already works.

For the record, a few things I checked and found correct, so they do not need re-litigating: the blockIdx index space against getRowGroups(), the queue/batch arithmetic across row-group boundaries (including the exact-multiple-of-capacity case), _metadata.row_index under splicing, subquery materialization timing, SQLMetric serialization, readBatch(num, vec, null, null) for top-level primitives, and ColumnIndexStore memoization across the three per-phase setRequestedSchema calls. I also initially suspected phase 2 was ignoring parquet.filter.columnindex.enabled, but that is a non-issue -- ColumnIndex and OffsetIndex are separate structures and phase 2 only consults the latter. A one-line comment at the phase-0 guard saying so would save the next reviewer the same detour.

dongjoon-hyun · 2026-09-22T14:34:39Z

+
+    WritableColumnVector[] dequeued = new WritableColumnVector[keyVectorQueues.length];
+    for (int i = 0; i < keyVectorQueues.length; i++) {
+      dequeued[i] = keyVectorQueues[i].removeFirst();


Leak: these vectors are unreachable by close() until line 632.

removeFirst() takes them out of keyVectorQueues, but pendingCloseKeyVectors = dequeued is only reached at the end of the method. If columnReader.readBatch(...) (L598) or cv.assemble() (L602) throws -- a corrupt page, an object-store read error -- the vectors are in neither the queues nor pendingCloseKeyVectors, so close() -> closeSplicingState() walks only the remaining queue entries and skips them.

With spark.sql.columnVector.offheap.enabled=true these are Platform.allocateMemory allocations freed only by releaseMemory() (no finalizer, no Cleaner), so each of the 4 task attempts leaks numKeys * capacity bytes permanently.

Moving pendingCloseKeyVectors = dequeued; to immediately after this loop closes it.

Fixed, and the shape changed rather than the assignment moving: a survivor vector now stays owned by its queue while the batch is built on it, and the next emit removes and closes it. So there is no window where a vector is out of the queues and not yet reachable from a field, and pendingCloseKeyVectors is gone.

dongjoon-hyun · 2026-09-22T14:34:39Z

+    List<BlockMetaData> blocks = lateMatReader.getRowGroups();
+    for (int blockIdx = 0; blockIdx < blocks.size(); blockIdx++) {
+      for (ColumnChunkMetaData chunk : blocks.get(blockIdx).getColumns()) {
+        if (projectedPaths.contains(chunk.getPath()) && chunk.getOffsetIndexReference() == null) {


This turns a performance feature into an availability one.

Any Parquet file without an offset index fails the query outright, with no per-file fallback. That is not an exotic shape: pyarrow's write_table defaults to write_page_index=False, Impala omits it, and parquet-mr < 1.11 (Spark <= 2.4) has no column/offset index at all.

Scenario: an operator sets spark.sql.parquet.storageFilterPushdown.enabled=true as a cluster default. Any join where InjectRuntimeFilter attaches a bloom to such a table now throws from initialize() on every task, retries 4x, and fails the stage -- where before this PR the bloom was an ordinary post-scan FilterExec and the query worked. One legacy file in one partition is enough.

Two things make it sting more than it needs to: the check is eager over every row group even when the filter keeps 100% of rows and phase 2 would never consult the index, and the only remedy the message offers is a global conf flip.

The reader already has precedent for degrading per file -- L731-741 sets storageFilter = null when every key column is missing. The same shape would work here: fall back to the eager read for this file and evaluate the predicate during splice. That keeps the "extraction already removed the conjunct" invariant intact while leaving the feature purely optional.

Narrowed rather than turned into a per-file fallback, and it came out a net deletion.

The up-front check is gone. Parquet resolves every requested column's offset index before it reads a byte, so phase 2's read is simply wrapped and MissingOffsetIndexException rethrown with the conf to set. That is narrower than the check was -- a row group the filter keeps whole never needs the index, since readFilteredRowGroup degrades to a plain read when the ranges cover the block, and one it rejects whole is never read at all -- so a file written without a page index still scans as long as the filter never prunes inside a row group. It is also more complete: an IO error while reading an index surfaces the same way, which a footer walk could not see.

The eager-read fallback does not work in this design: phase 1 has already buffered only the survivors, so a whole-block phase 2 would misalign the batch, and dropping the predicate is not an option because extraction has removed it from the post-scan Filter. What would remove the failure class entirely is a parquet-side change -- readFilteredRowGroup reading whole chunks while keeping the row ranges for getRowIndexes() -- and that is now named as a follow-up in the description.

dongjoon-hyun · 2026-09-22T14:34:39Z

+
+    long keyRowsTotal = pushedFilterRanges.rowCount();
+    PrimitiveIterator.OfLong rowIndexIter = pushedFilterRanges.iterator();
+    RowRanges.Builder finalRangesBuilder = RowRanges.builder();


Unbounded, and invisible to Spark's memory manager.

This loop evaluates the entire row group before loadNextRowGroupWithLateMaterialization returns, so keyVectorQueues holds every surviving key value of one row group at once -- and those vectors are live at the same time as phase 2's non-key vectors.

At the default 128MB parquet.block.size a row group is commonly 1M+ rows, and blooms are deliberately false-positive-prone, so most survive. For a 32-byte string key at capacity = 4096 that is ~250 full-width vectors plus their variable-length child buffers, per key column, per concurrent task. With off-heap enabled these go through Platform.allocateMemory rather than a MemoryConsumer, so the executor gets OOM-killed by YARN/k8s instead of raising SparkOutOfMemoryError or spilling.

The conf doc's "a task holds up to one extra copy of the key columns for one row group" is accurate but reads as bounded. It scales with row-group size, not with capacity, and for a dictionary-encoded key the accumulators store expanded bytes -- a chunk holding 1,000 distinct 40-byte values plus 1M dictionary IDs becomes ~20MB of materialized strings.

Evaluating one capacity-sized chunk at a time (emit, then continue) would bound this and fix the LIMIT stall below; a guard that falls back to the eager path above some blockRowCount would be the cheaper stopgap.

Bounded now, in the stopgap form you suggested, and the conf is spark.sql.parquet.storageFilterPushdown.maxSplicedRowGroupBytes (internal, 64MB). Phase 1 counts the bytes it has buffered and past the cap releases them and gives that row group up: phase 2 then reads every projected column of the surviving rows, so nothing is buffered and the cost is one extra read of the key columns. The count is examined once per accumulator, so the hot loop pays one add per key column and the check costs one comparison per batch worth of survivors.

Worth recording why it counts rather than estimates, since the first version did estimate from the footer. ColumnChunkMetaData.getTotalUncompressedSize() is the size of the uncompressed pages, which still hold dictionary ids: measured, a 2M-row BIGINT column with 1000 distinct values reports 1.26 bytes/row dictionary-encoded against the 9 the accumulator holds. Spark writes dictionary-encoded by default, so that estimate underestimated by 7x on the default path.

Chunk-at-a-time is discussed in the description. It would shorten what the first batch waits for, but the IO of a LIMIT query is unchanged either way, and chunking turns phase 2 into one read call per chunk and refetches any page that straddles a boundary.

dongjoon-hyun · 2026-09-22T14:34:39Z

+    return finalRangesBuilder.build();
+  }
+
+  private void ensureKeyScratchAllocated() {


If this allocation loop throws, close() cannot reach what it already allocated.

keyScratchBatch is assigned only after the loop completes, and close() (L338-342) frees the scratch vectors solely through keyScratchBatch. So an off-heap allocation failure on the second iteration leaves keyScratchVectors non-null holding a live OffHeapColumnVector while keyScratchBatch is still null -- close() tests keyScratchBatch != null, finds null, and skips them.

Secondarily, the if (keyScratchVectors != null) return; guard means a later call would proceed with a partially populated array and NPE at readers[i].readBatch(num, keyScratchVectors[i], ...).

Closing keyScratchVectors element-wise, the way closeSplicingState() already does for the accumulators, removes both.

Fixed: the array is assigned before the loop and close() frees the scratch vectors by walking it element-wise, so keyScratchBatch is no longer what reaches them.

The guard stays, and it cannot be reached in the state you describe: an allocation failure here propagates out of the reader, the task fails and close() runs, so nothing calls this again on a half-filled array.

dongjoon-hyun · 2026-09-22T14:34:39Z

+        cols[i] = persistentBatchColumns[i];
+      }
+    }
+    columnarBatch = new ColumnarBatch(cols);


This breaks resultBatch()'s documented contract.

resultBatch()'s javadoc says "Returns the ColumnarBatch object that will be used for all rows returned by this reader. This object is reused." Under splicing a new ColumnarBatch is created per batch, so a caller that follows the documented hoist-once pattern reads zero rows forever -- no exception, no wrong values, just silently empty.

That pattern exists in-tree: DataSourceReadBenchmark.scala:234 and ParquetEncodingSuite.scala:74 both do val batch = reader.resultBatch() outside the while (reader.nextBatch()) loop. Neither uses a storage filter today, and production v1 paths are safe because RecordReaderIterator.next() re-reads getCurrentValue per batch -- but resultBatch() is public and this suite's own helpers deliberately re-fetch it inside the loop, which suggests the constraint was already noticed.

Either update the javadoc, or keep one stable ColumnarBatch and mutate its column array in place.

Took the second option, since documenting an exception to a public method's contract is the worse half of the trade. There is one ColumnarBatch for the whole read again, and the emit path rewrites its key slots in place -- ColumnarBatch holds the column array by reference, its staging row included, so writing a slot publishes it. Two allocations per batch go away with it, and the hoist-once pattern works on this path.

dongjoon-hyun · 2026-09-22T14:34:40Z

+   * The per-emit batch is a transient view over vectors owned elsewhere; see {@link #close()} and
+   * {@link #closeSplicingState()}.
+   */
+  private boolean nextBatchSplicing() throws IOException {


Design question: why buffer the survivors instead of re-reading the key columns in phase 2?

This method, and most of the machinery it depends on, exists only because phase 2 deliberately omits the key columns. If phase 2 set requestedSchema instead of nonKeyRequestedSchema and read under finalRanges, the key columns would come back from the same PageReadStore, already aligned with the non-key ones, and the existing eager nextBatch() would serve every row group unchanged.

That would delete nextBatchSplicing (71 lines), ValueCopier + copierFor (64), initializeSplicingState / ensureKeyScratchAllocated / ensureCurrentKeyAccumulatorsAllocated / appendSurvivorRowToAccumulators / finalizePartialAccumulators / closeSplicingState (~103), the double-close handling in close(), the skipDataSlots parameter on allocateColumns, eight fields -- and ParquetStorageFilter.isSupportedKeyType with the type gate it forces into FileSourceStrategy.

It also removes, rather than fixes, several things flagged separately in this review: the whole-row-group buffering, the LIMIT stall, the per-row byte[] copy, and the "keyColumnIndices must stay sorted" coupling that the comment below warns can silently swap key columns in the output batch.

The honest cost is re-reading one narrow column's pages under a subset of the ranges phase 1 already read: always cheaper than phase 1 itself, usually still in page cache, worst case 2x key decode when the filter passes everything. That trades IO for the memory cost the design note at L218-224 already concedes, which seems like the easier direction to defend for a first cut.

If there is a reason this does not work, could it go in the design note? It is the first thing a reader of this method will ask.

Fair question, and it is now answered in the description under Design decisions as well as in the reader's own note.

Re-reading is where this started, and it was measured slower: on TPCDS q37, 3949 ms splicing against 4511 ms re-reading, with the feature off at 3199 ms. The 562 ms between the two is one more read of cs_item_sk, and nothing absorbs it on object storage -- parquet caches footers rather than pages, and S3A's default input stream caches nothing, so it is a new GET (the caching stream types are opt-in).

Two corrections to the cost model. The re-read does not remove the LIMIT stall, since phase 1 still walks the whole row group to build finalRanges. And it would not remove the buffering either, for the same reason -- what it removes is the splicing machinery, which is real but is the part with tests around it.

Your comment did change the planner, though. q37 is an all-keys projection, and on that shape neither design can win: the reader has to read the key column to evaluate the filter on it, so the scan reads the same columns for the same rows as a plain one. That is what 3199 against 3949 shows. Extraction is now dropped when every projected data column is a key column of the filter.

dongjoon-hyun · 2026-09-22T14:34:40Z

+   * unreachable. Teach both sides at once when adding a type; a type admitted there but missing
+   * here becomes a task failure instead of a planning-time rejection.
+   */
+  private static ValueCopier copierFor(DataType dt) {


This duplicates RowToColumnConverter, and has already drifted.

RowToColumnConverter.getConverterForType (sql/core/src/main/scala/org/apache/spark/sql/execution/Columnar.scala:288-327) has the identical groupings -- IntegerType | DateType | _: YearMonthIntervalType -> Int, LongType | TimestampType | TimestampNTZType | _: DayTimeIntervalType | _: TimeType -> Long, and the same MAX_INT_DIGITS / MAX_LONG_DIGITS three-way decimal split. BasicNullableTypeConverter also already handles the isNullAt -> putNull branch at L1249.

The hot loop already has an InternalRow in hand (keyScratchBatch.getRow(r)), so convert(row, currentKeyAccumulators) is close to a drop-in, and ColumnVectorUtils.java:134-138 is the existing precedent for calling this private[execution] Scala class from Java in this module.

The drift is visible in the next block: the VarcharType / CharType branches at L1367-1368 are dead, since both extend StringType (sql/api/.../CharType.scala:34, VarcharType.scala:33) and the StringType test comes first in the same || chain. ParquetStorageFilter.isSupportedKeyType's comment at L216 states this correctly, so the two files already disagree about the same fact -- which is the "keep them in lockstep" cost the javadoc above warns about.

The dead VarcharType and CharType branches are gone.

The dedup itself I would rather not do. RowToColumnConverter goes InternalRow to vector, while these copiers go vector to vector, so reusing it would put the row abstraction back in the innermost loop and, for a string, a UTF8String wrapper per value. That is the opposite direction from your appendBytes comment above, which asks the copy to get more specialized, and that one is the follow-up I would rather take.

dongjoon-hyun · 2026-09-22T14:34:40Z

+   * correct: the reader transfers nothing for them. Every caller for a given block walks the same
+   * metadata, so a skipped column drops out of the baseline and the per-phase totals alike.
+   */
+  private static long compressedBytesForRowRanges(


MessageType.getColumns() is rebuilt ~9x per row group, for two metrics.

getColumns() is not cached in parquet-mr (its source carries a // TODO: optimize this): each call rebuilds getPaths(0) and, per leaf, does getType(path) + getMaxRepetitionLevel + getMaxDefinitionLevel tree descents plus a new ColumnDescriptor.

ParquetFileReader.setRequestedSchema(MessageType) is setRequestedSchema(projection.getColumns()), so L961/L993/L1031 contribute three rebuilds per row group. This method calls schema.getColumns() twice per invocation (L1099, L1111) across three invocations (L986, L995, L1043) for six more. It also rebuilds a HashMap<ColumnPath, ColumnChunkMetaData> over every chunk of the block rather than the projected ones (L1104-1107), and rowRanges.rowCount() -- O(#ranges), and finalRanges can hold ~250K ranges for a 1M-row block at 50% selectivity -- is recomputed at L1009, L1099 and L1108.

For a 200-column projection on a 2000-column file that is roughly 2,400 allocations, 4,800 tree descents and 6,000 HashMap.puts per row group, to populate two counters.

Hoisting the three List<ColumnDescriptor> into fields in initializeLateMaterialization() (and using the setRequestedSchema(List<ColumnDescriptor>) overload), building the path->chunk map once per block, and passing the already-computed row counts in removes all of it with no behavior change.

Fixed exactly as suggested: the three List<ColumnDescriptor> are fields resolved once per file in initializeLateMaterialization, the reader is driven through the setRequestedSchema(List<ColumnDescriptor>) overload, the path-to-chunk map is built once per block and shared by the metric calls, and both row counts are passed in instead of recomputed.

dongjoon-hyun · 2026-09-22T14:34:40Z

+        "applied as an ordinary post-scan filter instead. Note that the surviving key values of " +
+        "a whole row group are buffered before the first batch of that row group is produced, " +
+        "so a task holds up to one extra copy of the key columns for one row group.")
+      .version("5.0.0")


Should this be 4.4.0?

dev/next_version_candidates.py currently prints master 5.0.0 / branch-4.x 4.4.0, and branch-4.x is at 4.4.0-SNAPSHOT and actively taking backports. This is an additive, opt-in, default-false feature, and the one public type whose signature changed (FileSourceScanExec, a new defaulted parameter) is under org.apache.spark.sql.execution.*, which project/MimaExcludes.scala:171 excludes permanently -- so neither master-only exception (binary-incompatible, or a non-critical dependency bump) seems to apply.

Unless you are deliberately planning this as master-only, 5.0.0 will make docs/sql-configuration and SHOW claim the config arrived a full major later than it did.

Master-only on purpose, so 5.0.0 is right. The reader needs the parquet 1.18 APIs -- RowRanges public in filter2.columnindex, RowRanges.builder(), addSelectedRow(long) and a public ParquetFileReader.getRowRanges(int) -- and that upgrade (#58120) went to master alone, with branch-4.x still on 1.17.0.

dongjoon-hyun · 2026-09-22T14:34:40Z

+    val sparkSession = fsRelation.sparkSession
+    val sqlConf = sparkSession.sessionState.conf
+    if (!sqlConf.parquetStorageFilterPushdownEnabled) return (Nil, afterScanFilters)
+    if (fsRelation.fileFormat.getClass != classOf[ParquetFileFormat]) {


Could the format answer this instead of the planner type-checking it?

This makes the generic FileSourceStrategy import ParquetFileFormat and ParquetStorageFilter (L36) and hard-code an exact-class check, so ORC or any third-party format can only participate by editing the planner, and ParquetFileFormat subclasses have no opt-in at all.

L205 goes a step further: ParquetStorageFilter.isSupportedKeyType is the set of types VectorizedParquetRecordReader.copierFor can copy -- a reader implementation detail the planner now owns, with a "must stay in lockstep" invariant spanning two packages and enforced only by comment.

Spark's established shape for this is a capability method on FileFormat, exactly like supportBatch, supportDataType, vectorTypes and metadataSchemaFields -- and supportBatch is already being called polymorphically five lines below. Something like:

def supportsStorageFilter(expr: Expression, readDataSchema: StructType): Boolean = false

would absorb both checks and keep the type list inside the parquet package.

The justification given above is that subclasses may override buildReaderWithPartitionValues -- but that concern is a consequence of this PR inverting the delegation, and it can be expressed inside the override, in the class that knows why: getClass == classOf[ParquetFileFormat] && ....

Done, as FileFormat.supportsStorageFilter(expr) defaulting to false. FileSourceStrategy no longer imports the parquet package, and the key-type list lives next to the copier that defines it. The subclass exclusion moved into ParquetFileFormat's override, where it reads as a fact about that reader, exactly as you put it.

One deviation: the signature takes only the expression, not the read schema. Everything the format needs is reachable from the expression's references, and schema-level feasibility is supportBatch, which the planner asks a few lines above. It can grow the parameter when a second format needs one.

peter-toth · 2026-09-23T12:30:00Z

Thanks @dongjoon-hyun for the review, I will adjust this PR and answer your findings soon, but let me move this to draft for now.

Addresses dongjoon-hyun's review of 2026-09-22. What changed, comment by comment: - Resource lifecycle. `pendingCloseKeyVectors` is gone: a survivor vector now stays owned by its queue while the batch is built on it, and the next batch closes it, so nothing can be left unreachable by `close()`. The phase-1 scratch vectors are closed through their array rather than through the batch that is assigned after the allocation loop, so a failed allocation cannot leak either. - `ParquetFileFormat`'s two public reader builders both route to a private `buildParquetReader`, so a subclass that overrides one and delegates to the other cannot recurse. The trait's scaladoc says so. - `resultBatch()` hands out one batch for the whole read again, storage filter or not. The emit path rewrites the batch's key slots in place instead of building a new `ColumnarBatch` per batch, which holds the method's documented contract for every caller and drops two allocations per batch. - The two unused `protected` fields are gone from `SpecificParquetRecordReaderBase`, and the footer is a local again. - Per-row-group work removed from the byte metrics: the three leaf-column lists are resolved once per file, the block's path-to-chunk map is built once per row group, and the row counts the caller already has are passed in instead of recomputed. - The dead `VarcharType` and `CharType` branches are gone from `copierFor`. - The test helpers drain inside `Utils.tryInitializeResource`, so a failure in the read loop closes the reader instead of leaking it. - The offset index is no longer checked up front. Phase 2's read is wrapped and parquet's own `MissingOffsetIndexException` rethrown with guidance. That is narrower, since a row group the filter keeps whole never needs the index, and more complete, since an IO error while reading an index surfaces the same way. It also made the change a net deletion. - New `UnsupportedFileReadException`, excluded from `DataSourceUtils.shouldIgnoreCorruptFileException` and thrown by every loud failure in this path. Without it `ignoreCorruptFiles` read those failures as a corrupt file and silently dropped the rest of a healthy file's rows. - The survivor buffer is bounded per row group by `spark.sql.parquet.storageFilterPushdown.maxSplicedRowGroupBytes` (internal, 64MB). Phase 1 counts the bytes it has buffered and gives that row group up once the count passes the cap, releasing what it holds; phase 2 then reads every projected column of the surviving rows, which buffers nothing and costs one extra read of the key columns. The count is examined once per accumulator, so it costs one comparison per batch worth of survivors rather than one per row. - `FileFormat.supportsStorageFilter(expr)`, defaulting to false, replaces the planner's exact-class check and its call into `ParquetStorageFilter`. `FileSourceStrategy` no longer imports the parquet package, and the reader's key-type list lives next to the copier that defines it. - Extraction is dropped when every projected data column is a key column of the filter. Such a scan reads the same columns for the same rows either way, since the reader has to read a key column to evaluate the filter on it, so pushing could only add the cost of evaluating the predicate outside the generated code. TPCDS q37 is that shape, and measured 3199 ms with the feature off against 3949 ms with it on. - `bytesAvoidedByPageFiltering` charges the second read of the key columns to the row group that gave splicing up. It credited the full non-key saving before and counted the extra read nowhere, so it reported a saving for a row group that transferred more bytes than a plain read would have. - `isSupportedStorageFilter` type-checks every reference of the conjunct, not only the ones on the bloom's value side, which is what `create` binds and re-checks. - Both confs declare a binding policy, `NOT_APPLICABLE`: neither can change how a view, UDF or procedure body resolves. - `.version("5.0.0")` stays. The feature needs the Parquet 1.18 upgrade, which landed on master only. - Comments and scaladoc across the change are trimmed to what the current code needs. Three comments are answered on the PR rather than in code. Chunk-at-a-time phase-1 evaluation is not worth it: parquet reads a row group's whole requested chunks before it hands back a page store, so the IO of a `LIMIT` query is the same either way and what the first batch waits for is one row group's key decode, while chunking would multiply the read calls and refetch pages that straddle a chunk boundary. The per-row `byte[]` in the string copier wants a vector-to-vector `appendBytes` on `WritableColumnVector`, which belongs in its own change. Reusing `RowToColumnConverter` for the copiers would put the row abstraction back in the hot loop, which is the opposite direction from that one. Verified: `ParquetStorageFilterSuite` 95 tests, plus `ParquetIOSuite`, `FileSourceStrategySuite`, `ParquetV1FilterSuite`, `ParquetV2FilterSuite`, `ParquetV1QuerySuite`, `ParquetV2QuerySuite`, `DataFrameJoinSuite` and `SubquerySuite`, 664 in all; `dev/lint-scala`; `dev/lint-java`.

peter-toth · 2026-09-23T15:36:42Z

Thank you for the review, @dongjoon-hyun -- it was unusually thorough, and two of the comments changed the design rather than the code.

Every comment is addressed, in one commit on top of the reviewed one (b9981cd) so the delta stays readable. Replies are inline. The larger ones:

The survivor buffer is bounded now, in the cheaper form you suggested: phase 1 counts the bytes it has buffered and gives that row group up past an internal cap, after which phase 2 reads every projected column of the surviving rows.
FileFormat.supportsStorageFilter replaces the planner's exact-class check, so FileSourceStrategy no longer imports the parquet package.
The offset-index check is gone. Phase 2's read is wrapped instead and parquet's own MissingOffsetIndexException rethrown with guidance, so a file without a page index still scans as long as the filter never prunes inside a row group.
Every loud failure on this path now throws a new UnsupportedFileReadException that ignoreCorruptFiles does not swallow. Chasing your offset-index comment turned that up: shouldIgnoreCorruptFileException matches any RuntimeException, so before this every failure the feature depends on could have been read as a corrupt file and silently dropped the rest of a healthy one's rows.
Your design question also produced a planner gate. An all-keys projection is no longer pushed at all: the reader has to read a key column to evaluate the filter on it, so such a scan reads the same columns for the same rows as a plain one and pushing can only add cost.

The description now carries a Design decisions section answering the design question up front, and a measured table on a real workload's file at the end of the test section.

dongjoon-hyun · 2026-09-23T16:37:46Z

+    }
+
+    val dataAttrs = AttributeSet(readDataColumns)
+    val (eligible, rest) = afterScanFilters.partition { expr =>


Extraction only checks deterministic, which changes the evaluation order. Before this PR, CombineFilters put the existing conjuncts first and the bloom was evaluated in FilterExec only on rows that passed them (short-circuit). Now phase 1 evaluates the bloom's value expression on every row in pushedFilterRanges.

Example: ANSI mode (the 4.x default), a string=bigint join key that InjectRuntimeFilter turns into might_contain(bf, xxhash64(CAST(s AS BIGINT))), plus a guard like WHERE kind = 'num' over rows whose s is not numeric. This succeeds with the conf off and fails with CAST_INVALID_INPUT with the conf on. evalAllMissing() has the same issue, since it evaluates the expression against the DEFAULT value.

Adding a throwable check alone won't catch this, because Cast doesn't override throwable.

Real, and reproduced rather than reasoned about: with the gate ablated, a CAST(s AS BIGINT) key plus a WHERE kind = 'num' guard over a non-numeric row in the same row group fails with CAST_INVALID_INPUT.

ParquetStorageFilter.isSupportedStorageFilter now requires the hash's children to pass ExprUtils.canEvaluateUnconditionally, which is the whitelist master already uses for this hazard class. That covers evalAllMissing too, since it evaluates the same expression. You were right about throwable: it is opt-in and Cast does not set it, which ExprUtils' own scaladoc calls out.

Tested on the gate and end to end, with the offending row in the same row group as matching ones so no statistic prunes it away. c255f88

dongjoon-hyun · 2026-09-23T16:37:46Z

+      keyScratchBatch.setNumRows(num);
+      for (int r = 0; r < num; r++) {
+        long blockRow = rowIndexIter.nextLong();
+        if (storageFilter.test(keyScratchBatch.getRow(r))) {


This evaluation now runs inside the file iterator, i.e. inside FileScanRDD's ignoreCorruptFiles catch. shouldIgnoreCorruptFileException excludes only UnsupportedFileReadException, so with spark.sql.files.ignoreCorruptFiles=true any evaluation error is treated as a corrupt file. An ANSI cast error from the key expression (see the comment on FileSourceStrategy) is enough. The rest of the file, including the not-yet-emitted survivors of the current row group, is then silently skipped and the query succeeds. Before this PR the same exception came from FilterExec, outside that catch, and failed the query. The existing ignoreCorruptFiles test covers only UnsupportedFileReadException.

Gone with the design change in the DataSourceUtils thread: there is no UnsupportedFileReadException and no case for it in the shared classifier any more.

The cast error you describe cannot reach the reader either, since such an expression is no longer offered. What is left is that an internal error in this reader is swallowed under ignoreCorruptFiles the way any other reader's is, which is now the same behaviour as the rest of the read path rather than a special case. c255f88

dongjoon-hyun · 2026-09-23T16:37:46Z

+      for (int r = 0; r < num; r++) {
+        long blockRow = rowIndexIter.nextLong();
+        if (storageFilter.test(keyScratchBatch.getRow(r))) {
+          finalRangesBuilder.addSelectedRow(blockRow);


finalRanges is built one row at a time, so scattered survivors produce roughly Np(1-p) ranges. In phase 2, every non-key leaf's VectorizedColumnReader builds a ParquetReadState, which walks getRowIndexes() and keeps its own List<RowRange> until the next row group. For a 1M-row row group with p=0.1..0.5 and 40 non-key leaves, that is about 140-380 MB per task, and maxSplicedRowGroupBytes doesn't bound it. Before this PR, row ranges came only from the column index at page granularity, so there were few of them.

Confirmed in the code: ParquetReadState.constructRanges materializes a List<RowRange> per leaf per row group, so scattered survivors cost about 40 bytes a range times the leaf count, and phase 2 drives one reader per leaf.

It is counted now and weighed against the same cap as the survivor buffer, in two rungs: drop the buffer first, and give the filter up for that row group if the ranges alone still do not fit. There is a second check right after phase 1, for a row group whose survivors fit in a single accumulator and are therefore never weighed inside the loop.

The better answer is a follow-up in the description: coarsen the ranges instead of giving the filter up. Closing a gap shorter than the smallest page row count of the columns phase 2 reads costs no extra bytes, since no page fits wholly inside such a gap. c255f88

dongjoon-hyun · 2026-09-23T16:37:46Z

+        // one it rejects whole is never read at all, so a file with no page index still scans as
+        // long as the filter never has to prune inside a row group.
+        try {
+          dataPages = lateMatReader.readFilteredRowGroup(blockIdx, finalRanges);


getColumnIndexStore memoizes one store per block, built at first call with the paths current then. Here that is init (getFilteredRecordCount) or phase 0, i.e. all requested columns. ColumnIndexStoreImpl.create returns EMPTY if any single column's offset index is missing, or if reading it throws an IOException (logged and treated as missing).

So when only a key column has the problem, this phase-2 read of non-key columns, which do have valid offset indexes, fails with "the file was written without a page index". The message is misleading, and ignoreCorruptFiles cannot skip the exception. The same file reads fine with no pushed data filter, or with the feature off.

Fixed by not letting it escape. The phase-2 read is wrapped, and on MissingOffsetIndexException the filter is given up for that row group and the read retried over pushedFilterRanges, which is what a plain scan reads. The file is latched, so later row groups skip phase 1 rather than evaluate a filter they cannot use.

So there is no exception left to be misleading, and nothing for ignoreCorruptFiles to swallow. The conf doc says what it costs: the first row group the reader tries it on also pays a key-column read, because a missing index is only reported by attempting the read. c255f88

dongjoon-hyun · 2026-09-23T16:37:46Z

+      SQLMetric rowsExcludedWithinRg = m.rowsExcludedWithinRowGroup();
+      if (rowsExcludedWithinRg != null && filteredRows > 0) rowsExcludedWithinRg.add(filteredRows);
+      if (bytesAvoidedPf != null) {
+        bytesAvoidedPf.add(nonKeyBaselineBytes - phase2Bytes);


SQLMetric.add ignores v < 0 (if (v >= 0) in SQLMetrics.scala). The negative contribution the comment above describes ("that is the truth about it") is therefore dropped silently, and bytesAvoidedByPageFiltering over-reports for row groups that gave up splicing.

Right, SQLMetrics.scala has if (v >= 0). The second key read is now charged into that row group's own phase-2 bytes, so the per-row-group number is honest, and the comment says what the clamp then does: a row group that read more than the baseline contributes nothing rather than subtracting.

A test asserts the charge, by running the same file and filter with the cap high and low and requiring the low arm to report less avoided. c255f88

dongjoon-hyun · 2026-09-23T16:37:46Z

+    }
+    // StringType covers CHAR and VARCHAR: both extend it.
+    if (dt instanceof StringType || dt instanceof BinaryType) {
+      return (dst, dRow, src, sRow) -> dst.putByteArray(dRow, src.getBinary(sRow));


On non-dictionary pages, src.getBinary(sRow) allocates a temporary byte[] for every surviving row, and the bytes are then copied a second time. High-cardinality string keys (UUIDs, for example) fall back to plain encoding, so this can mean millions of short-lived arrays per task. Could we copy directly from the source's byte storage by offset/length? The accumulators are also freshly allocated for every capacity-sized chunk and never pooled.

Follow-up in the description. The obvious shortcut does not work today: OffHeapColumnVector.getByteBuffer allocates a byte[] of its own, a dictionary-encoded source has to decode, and the two vector implementations disagree on whether the buffer position is absolute. Removing the copy needs the core column-vector classes, so it belongs in its own change.

Your point about plain-encoded high-cardinality keys is exactly right. The key-type tests now cover both encodings for real, with an assertion on what parquet actually wrote, which turned out to matter: the previous dictionary arm was writing PLAIN files.

dongjoon-hyun · 2026-09-23T16:37:46Z

+    val substitution = missingKeyLocalPositions.zip(missingKeyValues).toMap
+    val presentPositions = keyColumnIndices.indices.filterNot(substitution.contains)
+    val newPosOf = presentPositions.zipWithIndex.toMap
+    val rewritten = boundExpression.transform {


transform rebuilds BloomFilterMightContain via copy, which drops its @transient lazy val bloomFilter. For every file with a missing key column, the whole bloom (up to about 8 MB) is therefore deserialized again, and Predicate.create runs again. The substituted values come from the scan schema and are the same for every file, so memoizing per missing-key set within a task would avoid this.

Fixed: the rewrite is memoized per set of missing key positions for the task, so the bloom is deserialized once per scan rather than once per file. The substituted values are a function of those positions, since they come from the scan's schema, which is why the positions alone are a sound key. c255f88

dongjoon-hyun · 2026-09-23T16:37:46Z

+      }
+      storageFilter = storageFilter.rewriteForMissingKeys(missing, missingValues);
+
+      if (presentKeyColumns.isEmpty()) {


When every key column is missing and the rewritten predicate is false, the whole file is skipped without updating rowGroupsSkipped / rowsExcludedByRowGroup / bytesAvoidedByRowGroup. That doesn't match their documented meaning.

Fixed: the skip now walks every row group of the file and updates all three counters.

It measures against the rows the pushed data filter kept, which is the baseline every other skip path uses. Measuring against the whole row group would have let rowsExcludedByRowGroup exceed the reader's own row count, since that is getFilteredRecordCount(). A test asserts the numbers. c255f88

dongjoon-hyun · 2026-09-23T16:37:46Z

+      hadoopConf: Configuration,
+      storageFilterMetrics: Map[String, SQLMetric] = Map.empty
+    ): PartitionedFile => Iterator[InternalRow] = {
+    require(storageFilters.isEmpty,


The only caller, FileSourceScanExec.inputRDD, calls this only with non-empty storageFilters, which this require rejects. So the delegation to buildReaderWithPartitionValues below is unreachable. Yet it is exactly what creates the mutual-recursion trap the scaladoc warns third-party formats about. A default body that always throws, or a separate capability mixin with no default, would remove the trap.

Resolved from the other end by the model change: honoring a storage filter is optional, so this default is a legitimate implementation rather than dead code. It ignores the filters and delegates, which is correct precisely because the conjunct is still in the Filter, and the require is gone.

The recursion warning stays, with a corrected reason. The default delegates on this, so an override of buildReaderWithPartitionValues that delegates back here closes the loop. Reaching this default through super is part of that loop rather than an escape from it, which is what the old wording got wrong. c255f88

dongjoon-hyun · 2026-09-23T16:37:47Z

    .booleanConf
    .createWithDefault(true)

+  val PARQUET_STORAGE_FILTER_PUSHDOWN_ENABLED =


Nit: these confs (and their accessors near parquetFilterPushDown) split the spark.sql.parquet.filterPushdown group from its dependent sub-confs (.date, .timestamp, .decimal, .string.startsWith, .stringPredicate, inFilterThreshold). Storage-filter pushdown doesn't depend on filterPushdown, so could we move them after PARQUET_FILTER_PUSHDOWN_INFILTERTHRESHOLD?

Done, both confs and their accessors are after PARQUET_FILTER_PUSHDOWN_INFILTERTHRESHOLD now, so the filterPushdown group stays with its sub-confs. c255f88

dongjoon-hyun · 2026-09-23T16:39:21Z

Thank you for the quick update, @peter-toth. The review fixes addressed the previous round well.

I left another round of inline comments on b9981cda93b (review). Here is the summary.

Correctness (these seem worth addressing before merge)

Evaluation order: the extracted bloom is now evaluated before the remaining post-scan conjuncts. With ANSI mode, a key like CAST(s AS BIGINT) from InjectRuntimeFilter can throw on rows that an earlier guard used to filter out, so a query that works with the conf off fails with it on.
ignoreCorruptFiles: the same evaluation errors now happen inside FileScanRDD's catch, so they are treated as a corrupt file and the rest of the file is silently skipped.
Phase-2 memory: per-row finalRanges make each non-key column reader build its own RowRange list. That can reach hundreds of MB per task, and maxSplicedRowGroupBytes doesn't bound it.
Misleading failure: the per-block ColumnIndexStore is memoized over all columns, so a missing or unreadable offset index on one key column fails phase 2 with a "written without a page index" message.

Minor

bytesAvoidedByPageFiltering over-reports, because SQLMetric.add drops negative values.
The error message embeds the full bloom as hex.
Metrics are not updated when an all-missing-key file is skipped.

Design / cleanup

Since the runtime bloom is result-preserving, could it be an optional hint (fall back to the plain read) instead of must-honor with a new exception type?
The all-keys reader branch is unreachable in production, which contradicts the description.
Some things could be simplified: survivor-buffer memory accounting, reusing RowToColumnConverter, per-row byte[] copies, per-file bloom re-deserialization, the FileFormat default-builder recursion trap, and SQLConf placement.

- Keep the pushed conjunct in the post-scan Filter, so the scan's filtering is a copy rather than a transfer of responsibility - Cap the heap the phase-2 row ranges and the spliced key buffer take together, and give the filter up for a row group that would not fit - Retry a row group without late materialization when the file has no offset index - Push an expression only when it is safe to evaluate on every row of a row group - Cache the prepared filter's rewrites for files missing a key column - Remove UnsupportedFileReadException - Fold the splicing emit path into nextBatch and the batch release into one close body, and put every write and read helper of the suite on one - Measure a whole-file skip against the rows the pushed data filter kept, the baseline every other skip path uses, and skip the byte baseline a row group that already gave the filter up cannot report - Make the dictionary-encoded key-type tests write dictionary-encoded files, and assert the encoding they got - Cover what nothing asserted: key ordinals collected out of order, row groups that splice and row groups over the cap in both orders, the cap reached inside phase 1's survivor loop, the second key read in the byte metric, the rows a whole-file skip reports, and the row identity on the all-keys path - Correct the comments the copy model, the folded emit path and three wrong justifications left behind

peter-toth · 2026-09-24T18:41:03Z

Thank you for the second round. It was as thorough as the first, and the individual replies carry the details.

The comment that mattered most was the design question on treating the extracted filter as an optional hint. That is what the PR does now, and it went one step further than the suggestion: the conjunct is not moved out of the plan at all, it is copied. It stays in the post-scan Filter, so the reader may give a file or a row group up freely, and UnsupportedFileReadException, its case in the shared classifier and the hard failure on a page-index-less file are all gone.

Two things settled that direction. The bloom is already built by the time the scan runs, so having the Filter use it regardless costs nothing extra and means it is never built for nothing. And whoever turns this conf on is doing it because the filter prunes a lot, so evaluating it a second time on the rows that survive is a small price rather than a regression.

The description is rewritten around that, with a Design decisions section. It also now carries a measurement of splicing against re-reading the key columns, which answers your earlier question with numbers: re-reading transfers up to 49% more bytes than splicing, and more than the feature being off, because phase 1 has already read the key column once.

Everything is in c255f88.

dongjoon-hyun

Thank you for working on this, @peter-toth. I left some comments. I didn't find a wrong-result issue under the default settings. Most comments are about the escape hatch, the memory/range budgets, the missing-offset-index handling, the bloom filter shipping cost, and test coverage.

dongjoon-hyun · 2026-09-24T20:20:30Z

+        // applied wherever the filter keeps a row group whole (`readFilteredRowGroup` degrades to a
+        // plain read when the ranges cover the block) or rejects one whole.
+        try {
+          dataPages = lateMatReader.readFilteredRowGroup(blockIdx, finalRanges);


Phase 0 honors parquet.filter.columnindex.enabled=false, which is the documented escape hatch for a file with a wrong page index. However, this readFilteredRowGroup(blockIdx, finalRanges) still selects pages and synchronizes rows through the offset index whenever finalRanges is a strict subset of the block. In Parquet 1.18.1, readFilteredRowGroup(int, RowRanges) does not check options.useColumnIndexFilter(), unlike readNextFilteredRowGroup(), and VectorizedColumnReader takes pageFirstRowIndex from page.getFirstRowIndex().

So, for a file with a corrupt offset index, the plain path reads it correctly (with the escape hatch on, or with no pushed data filter at all, where it never touches the offset index). With this feature on, the phase-1 keys are still correct, but the phase-2 non-key values can come from the wrong rows. Since the post-scan Filter only re-checks the key hash, this becomes a silent wrong result. Could we skip late materialization (or read whole row groups in phase 2) when useColumnIndexFilter is false?

You are right, and the reasoning is exactly right too: the key the post-scan Filter sees is the correct one, so it cannot catch a non-key value taken from the wrong row.

Fixed by making the conf turn the whole path off rather than phase 0 alone. With parquet.filter.columnindex.enabled=false the reader declines the storage filter at init, so nothing reads part of a row group. The escape hatch is then total, which is what someone setting it means.

It also removes code: the phase-0 branch and the helper that held it are gone, and both sites now ask parquet for the ranges directly. The conf doc says the feature is off in that case. The test asserts the scan hands back every row of the file and reports all five metrics as zero, and that the same read with the conf on reports a saving, so it cannot pass by the feature being off everywhere. 2d8d39b

dongjoon-hyun · 2026-09-24T20:20:31Z

+    }
+    splicedBytes += keyFixedBytesPerRow + valueBytes;
+    currentKeyAccumulatorRowCount = dstRow + 1;
+    if (currentKeyAccumulatorRowCount == capacity) {


splicedBytes is updated per row, but it is compared against maxSplicedRowGroupBytes only here, when an accumulator reaches capacity rows. The post-phase-1 check (around line 1023) only looks at rowRangeStateBytes(survivorRangeCount), not at splicedBytes.

As a result, the advertised bound is not enforced:

A row group whose survivors fit in a single accumulator is never weighed. With 32KB string keys, 4095 survivors buffer about 128MB against the 64MB cap, and the buffer is kept through phase 2 and emit.

When survivors span multiple batches, the first check can already be one full accumulator past the cap.

In addition, the conf doc says such a row group "holds no more than the plain read path does", but during phase 1 the reader holds both the decoded capacity-sized keyScratchVectors (all rows) and the survivor accumulators, i.e., about twice the plain path's key memory, outside any MemoryConsumer. Could we check the cap per row (it is a single comparison) and include splicedBytes in the post-loop check, and update the doc?

Both real, and one owner weighs both halves now. The check runs after every surviving row, over splicedBytes + rowRangeStateBytes(survivorRangeCount), and there is no second check anywhere else. So a row group whose survivors fit in a single accumulator is weighed like any other, and crossing the cap cannot be an accumulator late.

Crossing it takes the cheaper concession first: release the buffer, and give the filter up as well only if the ranges alone still do not fit.

The doc claim is gone. It now says a task holds up to one extra copy of the key columns for one row group, and that what is counted is the buffered values and their per-row overhead, not the backing arrays a column vector may grow beyond that. @cloud-fan asked for the same on the range half, so this answers both. c29625f

dongjoon-hyun · 2026-09-24T20:20:31Z

+   * drives materializes the row group's range list of its own ({@code ParquetReadState}), and a
+   * filter whose survivors are scattered makes one range per surviving row.
+   */
+  private long rowRangeStateBytes(long rangeCount) {


This multiplies the range budget by the number of phase-2 leaf columns because ParquetReadState.constructRanges builds its own copy of the range list for every column reader. For a wide projection, the filter is given up at a small number of ranges: 64MB / (40B x 100 leaves) is about 16.8k ranges, which a 1M-row row group easily reaches with scattered survivors (e.g., the bloom filter's false positives).

The give-up happens only after phase 1 has already read, decoded, and evaluated the key columns, and then phase 2 re-reads the whole projection including the keys. The decision is not memoized per file (unlike fileHasNoOffsetIndex), so this repeats for every row group, which is slower than having the feature off. The conf doc's "as slow as not pushing the filter" understates this.

Since Parquet 1.18.1 exposes RowRanges.getRanges() and public Range.from/to, could ParquetReadState iterate one shared, immutable range list instead of copying it per column? That removes the x leaves factor at the root.

Taken at the root, and thank you for pointing at constructRanges rather than at the budget. ParquetReadState no longer builds a list at all: it coalesces the row indexes it walks into ranges lazily and holds the current one as two longs. The ranges are consumed once, in order, so the list bought nothing. Every read that uses a column index is lighter for it, not only this feature's.

With the x leaves factor gone, 64MB is about 1.6M ranges for one row group, and a row group cannot produce that many: the count is bounded by its row count, which at the default parquet.block.size is around 1M. So the case you describe stops being reachable, and what the budget still catches is the buffer, which is the half that really does grow with the key width. c29625f

dongjoon-hyun · 2026-09-24T20:20:31Z

+        // plain read when the ranges cover the block) or rejects one whole.
+        try {
+          dataPages = lateMatReader.readFilteredRowGroup(blockIdx, finalRanges);
+        } catch (MissingOffsetIndexException e) {


The absence of an offset index is known from the in-memory footer without any IO: ColumnChunkMetaData.getOffsetIndexReference() == null, which is exactly when ParquetFileReader.readOffsetIndex returns null and ColumnIndexStoreImpl throws MissingOffsetIndexException. Detecting it only here means:

For every split of a file written without a page index (e.g., pyarrow's default write_page_index=False), the first partially-surviving row group pays the phase-1 buffering, a WARN log with a stack trace, and a full re-read of the projection including the keys.

After that, filterGivenUp = fileHasNoOffsetIndex skips phase 1 for the rest of the split, so even whole-row-group skips, which don't need an offset index, are lost.

With column-index filtering on and a pushed filter, a missing offset index on only a key column makes the init-time ColumnIndexStore for the block EMPTY, so the phase-2 read of non-key columns also throws and the filter is given up for the whole file, even though the non-key columns have offset indexes.

Also, the SQLConf doc sentence "since a missing page index is only reported by attempting the read" is not accurate. Could we check the footer up front instead?

Three of the four are fixed. The footer check is the one I am not taking, and the reason is this PR's own history.

Fixed: phase 1 keeps running on such a file. fileHasNoOffsetIndex now only stops the buffering, so the row groups the filter empties are still skipped whole for the rest of the split. The mixed case reads plainly instead of giving the file up, since the filter is given up for that row group and the read retried over pushedFilterRanges. The WARN says what happened and what still works. The SQLConf sentence you quote is gone.

The footer walk I did write, and then removed: round 2 of this PR had already deleted the same walk (2cb98b2), for two defects mine had again.

Over-strict. Parquet resolves the index over the paths current at its first lookup for the block, which with no pushed data filter is phase 2's non-key columns alone. A walk over the whole projection therefore rejects a file whose key column has no offset index, where the read would have succeeded.

Incomplete. An IOException while reading an offset index raises the same exception, and the footer cannot see that.

What it would save is one row group's phase-1 buffering per split and one stack trace, because the throw lands before any data page is read. That is worth less than a check that is wrong in both directions, so the exception stays the source of truth. c29625f

dongjoon-hyun · 2026-09-24T20:20:31Z

+      // the first `requiredSchema.length` attributes line up with its fields.
+      val requestedDataAttrs = output.take(requiredSchema.length)
+      storageFilters.map { expr =>
+        val subqueryReplaced = expr.transform {  case s: execution.ScalarSubquery => s.toLiteral }


This embeds the materialized bloom filter (up to 8MiB with the default spark.sql.optimizer.runtime.bloomFilter.maxNumBits) as a binary Literal into ParquetStorageFilter, which is captured by the reader closure of FileScanRDD. Since it is a distinct object from the copy the post-scan Filter already ships, the stage task binary grows by the bloom size, and each task deserializes the literal and then deserializes it again into a BloomFilter via Predicate.create (and once more per rewriteForMissingKeys). That is roughly 16MiB of extra heap per concurrent task. Could we broadcast the bloom bytes once (like InSubqueryExec.resultBroadcast) and share the deserialized filter per executor?

Follow-up, now in the description: broadcast the bytes once and share the deserialized filter per executor, the way InSubqueryExec.resultBroadcast does.

The repeated deserialization per task is already down to one. The rewrite for missing key columns is memoized per set of missing positions for the task, so rewriteForMissingKeys no longer rebuilds the filter per file. What is left is the second copy in the task binary, and that is inherent to the conjunct staying in the plan, so removing it is the broadcast rather than a local fix. c29625f

dongjoon-hyun · 2026-09-24T20:20:31Z

+    }
+    long blockRowCount = reader.getRowGroups().get(blockIndex).getRowCount();
+    boolean wholeBlock = rowRangeCount == blockRowCount;
+    ColumnIndexStore ciStore = wholeBlock ? null : reader.getColumnIndexStore(blockIndex);


The correctness here relies on an implicit ordering invariant: Parquet caches the ColumnIndexStore per block with the paths at the time of the first call, and setRequestedSchema does not invalidate it. For a path outside the cached set, MISSING_INDEX_STORE.getOffsetIndex() returns null rather than throwing. I verified that the current phase ordering never builds the store with a narrower path set than a later lookup, but a future change that calls this before phase 2 (e.g., to estimate bytes for gating) would make the subsequent read NPE inside Parquet's filterOffsetIndex(null, ...) (not the MissingOffsetIndexException this PR catches) and would silently undercount the metrics here. Could we snapshot the per-block ranges/stores with the full projection at initialization, so the invariant is enforced by code rather than by comments?

The invariant is enforced where it is created rather than snapshotted. Phase 0 sets the full projection before it asks for the ranges, and that is the first lookup for the block, so every later lookup is over a subset. With a pushed data filter getFilteredRecordCount() at initialize gets there first, also with the full projection. A snapshot would not remove that requirement, it would move it: something still has to build the store with the full projection before phase 1 narrows the schema, which is what phase 0 does.

The metric walk no longer has a stake in it either. A path the store was not built with is skipped and reported once per file as an undercount, never thrown. I did write the throw first, and then removed it: ignoreCorruptFiles turns any exception from a reader into a silently truncated file, so a byte counter must not be able to change the answer. c29625f

dongjoon-hyun · 2026-09-24T20:20:31Z

+      outputDataSchema: StructType): Seq[Expression] = {
+    val sparkSession = fsRelation.sparkSession
+    val sqlConf = sparkSession.sessionState.conf
+    if (!sqlConf.parquetStorageFilterPushdownEnabled) return Nil


The doc of this method says it names neither a format nor a type, but it reads the Parquet-named conf spark.sql.parquet.storageFilterPushdown.enabled in the format-agnostic strategy. Similarly, the generic FileSourceScanLike defines five Parquet-specific metrics (row groups, page filtering), and ParquetFileFormat imports the physical node's companion object to get their keys. Once a second format implements supportsStorageFilter (or the planned DSv2 step lands), it would be switched by a Parquet conf. Could we move the conf check into ParquetFileFormat.supportsStorageFilter (as supportBatch already reads Parquet confs inside the format) and let the format declare its metrics?

The conf moved: FileFormat.supportsStorageFilterPushdown(sparkSession), default false, and ParquetFileFormat reads the conf there together with its exact-class check, the way supportBatch reads Parquet confs inside the format. FileSourceStrategy asks it once per scan, before anything per conjunct, and names no format.

The five metrics stay for now, as a follow-up in the description. Moving the constants alone would leave the same Parquet vocabulary in FileSourceScanLike, only indirectly. What makes it right is a format declaring its own metrics, and the shape of that is decided by the second format to implement this. c29625f

dongjoon-hyun · 2026-09-24T20:20:31Z

+   * Closes anything held by the splicing path: every vector still queued, the published head
+   * included, and partially-filled accumulators. Called from {@link #close()}.
+   */
+  private void closeSplicingState() {


closeSplicingState re-implements almost the same body as abandonSplicing, and the push-to-queue loop is duplicated between appendSurvivorRowToAccumulators and finalizePartialAccumulators. Similarly, phase 0's useColumnIndexFilter ? getRowRanges(...) : createSingle(...) logic with the empty-block/zero-row guards is copied into recordFileSkipped. Since these handle off-heap vector lifetimes and the metric baseline, changing only one copy could lead to double-free/leak or inconsistent metrics. Could closeSplicingState call abandonSplicing(), and could the range computation be one helper such as pushedFilterRanges(blockIdx)?

closeSplicingState calls abandonSplicing now, and the push-to-queue loop is one pushAccumulatorsToQueues that both callers go through.

The range duplication went the other way. Your first comment turned parquet.filter.columnindex.enabled=false into a gate on the whole feature, so the branch is gone and both sites are one getRowRanges(blockIdx) call, which is why there is no helper. What is left duplicated is the two-line empty-block guard, and I left that alone because each site does something different after it. 2d8d39b

dongjoon-hyun · 2026-09-24T20:20:31Z

+          if (blockRow != previousSurvivor + 1) survivorRangeCount++;
+          previousSurvivor = blockRow;
+          if (accumulate) {
+            accumulate = appendSurvivorRowToAccumulators(r);


The same "give up" state is signaled through several channels: the boolean return of appendSurvivorRowToAccumulators and the local accumulate always equal spliceCurrentRowGroup; evaluateStorageFilter signals the filter give-up both via the filterGivenUp field and a null return, but the caller never checks for null; and survivorRangeCount is an implicit out-parameter between methods. Since (filterGivenUp, spliceCurrentRowGroup) has only three legal states (splice, filter-only, plain), could we model it as a single enum and let the caller check survivors == null?

Two of the three channels are gone. appendSurvivorRowToAccumulators returns the bytes it added rather than a flag, the local accumulate with it, and the caller now checks survivors == null for the filter give-up. survivorRangeCount is a local of the loop that owns the budget.

The enum I did not do, because the two booleans are not redundant. filterGivenUp and !spliceCurrentRowGroup name different states and the byte metric reads them separately: a row group that gave splicing up still reports a saving, with its second key read charged against it, while one whose filter was given up reports nothing, since it read what a plain scan reads. The transitions are giveUpFilter() and abandonSplicing(), and the first calls the second, so the fourth combination cannot be built. c29625f

dongjoon-hyun · 2026-09-24T20:20:31Z

+          FileSourceScanLike.STORAGE_FILTER_BYTES_AVOIDED_BY_ROW_GROUP, null),
+        bytesAvoidedByPageFiltering = storageFilterMetrics.getOrElse(
+          FileSourceScanLike.STORAGE_FILTER_BYTES_AVOIDED_BY_PAGE_FILTERING, null))
+      // `create` requires every condition extractStorageFilters already pre-checked, so a violation


nit. extractStorageFilters doesn't exist. It seems to have been renamed to FileSourceStrategy.storageFiltersFor. The same stale reference is in ParquetStorageFilter.scala (line 150) and ParquetStorageFilterSuite.scala (line 339). In addition, the test comments at lines 285 (pendingCloseKeyVectors) and 1416 ("ColumnarBatch is replaced on every nextBatch()") describe an older design, while the current reader reuses one batch object and rewrites its key slots in place.

All fixed: storageFiltersFor in the three places, and the two test comments now describe the current design. The batch is one object whose key slots are rewritten in place, and resultBatch()'s javadoc says that, since the old comment was what round 1 flagged as a contract break. @cloud-fan flagged the same comment, so this answers both. c29625f

dongjoon-hyun · 2026-09-24T20:21:29Z

Here is a summary of my review comments (#58895 (review)) on c255f88. I didn't find a wrong-result issue under the default settings.

Correctness

The parquet.filter.columnindex.enabled=false escape hatch only covers phase 0. Phase 2 still selects pages and synchronizes rows through the offset index, so a file with a corrupt offset index can return non-key values from the wrong rows while the post-scan Filter passes them (key hash is correct).
maxSplicedRowGroupBytes is checked only when an accumulator fills capacity rows, and the post-phase-1 check ignores splicedBytes. Large variable-length keys can exceed the cap by a wide margin, and the conf doc ("holds no more than the plain read path does") is not accurate.
The key-slot/queue pairing in publishSurvivorKeyVectors silently depends on ParquetStorageFilter.create sorting the ordinals.
FileFormat.supportsStorageFilter receives canonicalized expressions because ExpressionSet.filter applies the predicate to e.canonicalized (undocumented contract for third-party formats).

Performance

The row-range budget is multiplied by the number of phase-2 leaf columns (per-reader copies in ParquetReadState). Wide projections give the filter up after phase 1 has done all its work, repeatedly for every row group, which is slower than the feature off. Sharing RowRanges.getRanges() would fix it at the root.
A missing offset index is detected only via MissingOffsetIndexException in phase 2, although it is known from the footer (getOffsetIndexReference() == null). This costs wasted work and a stack-trace WARN per split, loses whole-row-group skips afterwards, and gives up the whole file when only a key column lacks an offset index.
The materialized bloom filter is duplicated into the task binary and deserialized twice per task (about 16MiB extra heap per concurrent task with the default max size).
Because Cast is not in the canEvaluateUnconditionally whitelist, even never-failing widening casts (e.g., int joined with bigint) are never pushed.

Design

The correctness of the byte metrics and reads relies on the implicit ordering of Parquet's per-block ColumnIndexStore cache (MISSING_INDEX_STORE.getOffsetIndex() returns null, which would NPE inside Parquet if the ordering changes).
A Parquet-named conf and Parquet-specific metrics live in the format-agnostic FileSourceStrategy / FileSourceScanLike.

Tests

The >= 0 metric assertions can never fail because SQLMetric ignores negative values.
The MissingOffsetIndexException path is untested, although such a file can be written with the ParquetFileWriter.writeDataPage overloads without a row count.
No AQE-on test verifies that the storage filter is actually applied (only result equality, which the post-scan Filter guarantees anyway).

Cleanup

Duplicated code (closeSplicingState vs abandonSplicing, phase-0 ranges in recordFileSkipped) and redundant give-up signaling (filterGivenUp, spliceCurrentRowGroup, accumulate, null return).
Stale references to the nonexistent extractStorageFilters and outdated test comments.

cloud-fan

Review summary

The row-group survivor loop must own the combined key-value and range budget. Partial accumulators and later filter-only range growth currently escape enforcement, and the range-growth path can exceed the configured bound before fallback, so this blocks landing.

The phase-1 key read also needs to participate in the missing-offset-index fallback so this optional optimization cannot fail an otherwise readable scan. Separately, its consumed key-only PageReadStore needs lexical ownership and closure. The partial-accumulator cap issue and stale test comment are already covered by existing review threads.

Findings

5 total: 0 P0, 2 P1, 2 P2, 1 P3.

Blocking (P1)

Apply missing-offset-index fallback to the phase-1 key read — sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:1013 — see inline.
Keep enforcing the range budget after splicing is abandoned — sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:1325 — see inline.

Non-blocking (P2)

Close the phase-1 page store at its acquisition boundary — sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:1013 — see inline.
Enforce the splice cap for partial accumulators — sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:1404 — unresolved in existing discussion.

Nit (P3)

Update the stale survivor-vector lifecycle comment — sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilterSuite.scala:285 — unresolved in existing discussion.

Existing discussions

existing discussion — The feedback identifies missing-key offset-index handling and late detection, while the local finding establishes a distinct earlier phase-1 exception that bypasses the only fallback catch.
existing discussion — The thread establishes that missing-index fallback has no regression-sensitive test; the local finding adds the uncovered phase-1 mixed-index failure point.

cloud-fan · 2026-09-25T10:57:01Z

+      long finalRowCount = baselineRows;
+      if (!filterGivenUp) {
+        lateMatReader.setRequestedSchema(keyOnlyColumns);
+        PageReadStore keyPages = lateMatReader.readFilteredRowGroup(blockIdx, pushedFilterRanges);


Non-blocking (P2): Please close this phase-1 PageReadStore after evaluateStorageFilter finishes, including skip, give-up, and exception exits. readFilteredRowGroup returns an AutoCloseable store whose page readers and ByteBufferReleaser otherwise stay live; this path acquires one extra store for every attempted row group, so long scans can accumulate direct-memory pressure. A try-with-resources around the phase-1 acquisition and evaluation matches the ownership boundary.

Done: try-with-resources around the phase-1 acquisition and evaluation, so the store is closed on every exit, including skip, give-up and exception.

The ownership point is right and worth writing down, which the code now does: readFilteredRowGroup hands out a store the file reader does not track, unlike readNextRowGroup, so nothing else would close it. Phase 2's store is closed by this reader too, at the top of the next row group's load and in close(). c29625f

cloud-fan · 2026-09-25T10:57:01Z

+      long finalRowCount = baselineRows;
+      if (!filterGivenUp) {
+        lateMatReader.setRequestedSchema(keyOnlyColumns);
+        PageReadStore keyPages = lateMatReader.readFilteredRowGroup(blockIdx, pushedFilterRanges);


Blocking (P1): This key-only read is the first offset-index-dependent operation, but it sits outside the MissingOffsetIndexException fallback used for phase 2. A valid mixed-index file can have pushedFilterRanges narrowed by one indexed column while this key column lacks an offset index; enabling this optional optimization then fails a query the plain reader accepts. Please extend or preflight the fallback at this phase-1 boundary so the row group is read plainly.

This one cannot arise, and there is now a test for the file that would produce it.

Phase 0 sets the full projection before it asks parquet for the ranges, so the block's ColumnIndexStore is built over every projected column, key columns included. ColumnIndexStoreImpl.create returns EMPTY when any of them has no offset index, and with an empty store ColumnIndexFilter narrows nothing, so pushedFilterRanges covers the whole block. readFilteredRowGroup(int, RowRanges) then short-circuits to internalReadRowGroup and consults no index at all. The mixed-index file therefore reaches phase 1 with whole-block ranges: one column without an offset index degrades the whole block, which is what makes the phase-1 read safe.

The new test writes a file with no offset index for any column and asserts phase 1 still runs on it. The phase-2 boundary keeps the catch, because that is where a strict subset is asked for, and the retry there no longer trusts the argument above: if pushedFilterRanges were ever a strict subset on such a file, it fails with a message that says so rather than throwing from inside parquet. c29625f

cloud-fan · 2026-09-25T10:57:01Z

+          if (blockRow != previousSurvivor + 1) survivorRangeCount++;
+          previousSurvivor = blockRow;
+          if (accumulate) {
+            accumulate = appendSurvivorRowToAccumulators(r);


Blocking (P1): Once this returns false, evaluateStorageFilter still adds every later match to finalRangesBuilder, but the per-survivor budget check is disabled with accumulation and the remaining check happens only after the full row group. Wide keys can therefore give up splicing early and still build an unbounded scattered range list, violating the configured per-row-group memory cap. Please make the survivor transition own both key-byte and range accounting, including after splicing is abandoned, and fall back as soon as the combined budget is crossed.

Recommended change: Centralize the combined spliced-value and row-range budget check in the per-survivor loop, apply it to partial accumulators and filter-only mode, and stop retaining survivor state immediately when the bound is exceeded. Align the config documentation and add boundary scenarios for large values and post-give-up scattered ranges.

Why this works: After each survivor changes retained value or range state, compute the same combined accounting regardless of accumulator fullness or spliceCurrentRowGroup. If the next retained state exceeds the cap, release key accumulators, abandon the storage filter for that row group, and avoid constructing the remainder of the survivor range list.

Scope: Make one row-group survivor-state owner enforce the advertised memory bound in every accumulation mode.

Compatibility: Eligible scans may still splice within budget, while every give-up path continues to behave as an optional optimization and preserves query results.

Risks: Giving up in the middle of evaluation must not leave partial key vectors or ranges observable by emit. Checking before versus after adding a survivor must use a consistent inclusive cap boundary.

Constraints: Fallback must keep every row selected by pushedFilterRanges because the post-scan conjunct remains authoritative. The accounting may acknowledge column-vector capacity slack, but it must not claim a hard bound it does not enforce.

Success: A partial accumulator of large variable-width keys cannot remain retained beyond the configured combined budget. Once splicing is abandoned, later scattered survivors cannot grow range state beyond the same budget. Budget give-up returns the same rows as a plain read and releases partial survivor vectors.

Done as described. One check, in the per-survivor loop, over splicedBytes + rowRangeStateBytes(survivorRangeCount), and it runs whether or not splicing is still on. Crossing it releases the key accumulators first, since that is the cheaper concession, and gives the filter up when the ranges alone exceed the cap, which returns immediately so the rest of the range list is never built. Nothing partial is left observable: giveUpFilter closes every buffered vector, and the caller then reads the row group over pushedFilterRanges, which is what a plain read reads.

The conf doc is aligned with it: two pools, both growing with the survivor count, examined after every surviving row, and what is counted is the values and their per-row overhead rather than the backing arrays.

Tests cover both boundaries, a row group that crosses the cap inside the loop and one whose ranges alone do, plus a file whose first row group splices while a later one does not, since the batch is one object and its key slots have to go back to the vectors phase 2 reads into.

The range half also got much cheaper at the root, from @dongjoon-hyun's comment on the same loop: ParquetReadState no longer copies the range list per column reader. c29625f

dongjoon-hyun's third round and cloud-fan's review. - Fail open instead of declining a conjunct the reader might not be able to evaluate on every row: an error that carries an error class and is not internal gives the filter up for that row group, and the post-scan Filter then evaluates the conjuncts in their own order. Requiring a total expression also declined every widening cast, which is what type coercion inserts for an int-to-bigint join - Coalesce row ranges lazily in ParquetReadState, and hold the current range as two longs, so one range at a time is held per column reader instead of a list each. That is what made the memory cap reachable, and every column-index read is lighter for it - Weigh that cap after every surviving row, over the buffered key bytes and the row ranges together, so a row group whose survivors fit in one accumulator and a range list that grows after the buffer is gone are both bounded. The check after phase 1 goes with it, the loop catches everything now - Keep applying the filter to a file with no offset index wherever it empties a row group, rather than giving the whole file up - Close the page stores phase 1 and phase 2 read from - Pair key slots with queues through keyColumnIndices, so the emit path does not depend on the ordinals being sorted - Ask the format once per scan whether it applies storage filters at all, which is where the conf that enables them belongs, and offer the original conjuncts rather than their canonicalized form - Enforce the column-index-store ordering the byte metrics rely on, with the footer rather than a comment - Cover what nothing asserted: a file with no offset index, an unsorted ordinal collection, the cap crossed inside the survivor loop, the metrics of a whole-file skip, and that AQE keeps the filter on the scan - Replace three metric assertions that could not fail, and correct the stale names and comments two reviewers found

parquet.filter.columnindex.enabled=false says the file's page index is not to be trusted. Phase 0 honoured it, but phase 2 reads part of a row group through the offset index whatever the conf says, so a wrong index could pair a row's key with another row's values. The reader now declines the filter outright, which also removes the phase-0 branch and its helper.

peter-toth · 2026-09-25T17:13:20Z

Thank you both. @dongjoon-hyun's third round found the one path in this change that could return a wrong answer, and @cloud-fan's blocking comment landed on the same survivor loop from the other side.

The change to look at first: parquet.filter.columnindex.enabled=false now turns this feature off entirely. Phase 0 honoured it, which was a half measure, because phase 2 reads part of a row group through the offset index whatever that conf says. A wrong index there pairs a row's key with another row's values, and the post-scan Filter cannot catch it, since the key it sees is the right one. The escape hatch is total now.

The rest of the round:

One owner weighs the memory budget, after every surviving row, over the buffered key bytes and the row ranges together. A row group whose survivors fit in a single accumulator is weighed like any other, and the range half is far cheaper at the root: ParquetReadState coalesces lazily instead of materializing a list per column reader, so the x leaves factor is gone.
The reader fails open on an evaluation error instead of the planner rejecting cast keys, so an int joined with bigint is pushed again.
The phase-1 page store is closed at its own boundary, the key-slot pairing is read off the ordinals, the format is asked about the original expression rather than a canonicalized one, and the byte-metric walk reports an undercount instead of throwing, because ignoreCorruptFiles would turn a throw into a silently truncated file.
Tests for what was untested: a file with no offset index at all, written with the low-level writer overloads you named, and an AQE arm that asserts the final plan's scan still carries the filter.

Three things are follow-ups in the description rather than in this PR: broadcasting the bloom bytes, letting a format declare its own metrics, and coarsening the survivor ranges instead of giving the filter up. One is a reasoned no, in its own thread: the offset-index footer preflight, which round 2 of this PR had already deleted for two reasons my new attempt had again.

The description is restructured, with the phases and the six ways the reader stops short up front. The suite is at 99 tests.

dongjoon-hyun

Thank you for working on this, @peter-toth. I left 15 inline comments. Here is a summary, roughly ordered by severity.

Correctness

VectorizedParquetRecordReader L1357: fail-open covers only non-internal SparkThrowables. When a built-in key expression (e.g. timestamp_seconds(decimal), format_string) throws a raw JDK exception on a row that an earlier conjunct would reject, the reader rethrows it. The query then fails where the plain plan succeeds, or, with spark.sql.files.ignoreCorruptFiles=true, the rest of the file is silently dropped.
VectorizedParquetRecordReader L1085: phase 2 trusts the Parquet OffsetIndex even for bloom-only queries, where the plain path never consults it. A present-but-wrong offset index would pair keys with the wrong values.
VectorizedParquetRecordReader L1378: the survivor buffer is outside any MemoryConsumer, and the budget undercounts it by about 2-4x for string keys.
VectorizedParquetRecordReader L604: a data column named _tmp_metadata_row_index can become a storage-filter key.
VectorizedParquetRecordReader L1026: under ignoreCorruptFiles, a corrupt key page drops the earlier rows of its row group.
ParquetFileFormat L216: the subclass guard is missing from buildReaderWithStorageFilters. This affects extensions only.

Design / efficiency / cleanup

L666: swapping key vectors into the batch and freeing them, vs. copying into the persistent vectors.
L712: parquet.filter.columnindex.enabled=false could use the existing no-index degraded mode instead of disabling the feature.
L1052: phase-1 work is wasted on files without an offset index.
L1369: there is no density check for scattered survivors.
L773: the missing-key predicate rewrite duplicates ParquetColumnVector's missing-column rule.
L1087: a WARN with a stack trace is logged per file for no-offset-index files.
L1064: the all-keys mode is unreachable in production.
ParquetStorageFilter L50: nullable StorageFilterMetrics are used only by tests.
DataSourceScanExec L661: Parquet-specific metrics are defined in FileSourceScanLike.

Item 1 is the most important one, since it can silently drop rows.

dongjoon-hyun · 2026-09-25T21:15:22Z

+        try {
+          survives = storageFilter.test(keyScratchBatch.getRow(r));
+        } catch (RuntimeException e) {
+          if (!storageFilter.isEvaluationError(e)) throw e;


Fail-open covers only non-internal SparkThrowables. isEvaluationError (ParquetStorageFilter.scala:157) returns false for any plain JDK RuntimeException, so the exception is rethrown here, and also from evalAllMissing during initialize. Some built-in key expressions admitted by InjectRuntimeFilter.isSimpleExpression and canEvaluateInTheReader throw raw JDK exceptions, e.g. timestamp_seconds(decimal_col) (ArithmeticException from longValueExact) or format_string('%c', cp) (IllegalFormatCodePointException).

Take ... JOIN d ON timestamp_seconds(a.d) = d.ts WHERE a.kind = 'num', where the offending rows have kind = 'text'. The pushed filter prunes only at page granularity, so phase 1 evaluates the bloom on those rows too:

spark.sql.files.ignoreCorruptFiles=false: the task fails, while the same query succeeds with the feature off. That contradicts the contract documented on FileFormat.buildReaderWithStorageFilters ("A reader must not fail the query for that").

spark.sql.files.ignoreCorruptFiles=true: FileScanRDD treats the RuntimeException as a corrupt file and silently skips the rest of the file, so join rows go missing. Without the feature, the same exception would come from FilterExec, outside that try, and fail the query.

Since the conjunct stays in the post-scan Filter, failing open on any non-fatal exception from test()/evalAllMissing() looks safe. A genuine error is raised again by FilterExec on the same row, in plan order.

dongjoon-hyun · 2026-09-25T21:15:22Z

+        // read and blind to an index that is claimed but unreadable. It is also where the throw
+        // costs least: it lands before any data page is read.
+        try {
+          dataPages = lateMatReader.readFilteredRowGroup(blockIdx, finalRanges);


This extends trust in the Parquet OffsetIndex to bloom-only queries. With no pushed data filter, or one that prunes no page, the plain read path never consults the offset index, because readNextFilteredRowGroup falls back to readNextRowGroup. Here, phase 2 reads the non-key columns through internalReadFilteredRowGroup for every row group the bloom narrows, and the page firstRowIndex values come from the offset index.

Consider a file whose offset index is present but wrong. An example is a writer that split repeated records across pages while writing a page index (apache/arrow-rs#3680). Such a file is read correctly today, but with this feature the non-key values would be paired with the wrong spliced keys. The key itself is correct, so the post-scan Filter cannot catch it. The only escape hatch is parquet.filter.columnindex.enabled=false, and a user with no pushed predicate has no reason to set it. It would be good to at least document this in the conf doc.

dongjoon-hyun · 2026-09-25T21:15:22Z

+          // concession comes first: release the buffer, and give the filter up as well if the
+          // ranges alone still do not fit.
+          long rangeBytes = rowRangeStateBytes(survivorRangeCount);
+          if (splicedBytes + rangeBytes > cap) {


The survivor buffer is outside any MemoryConsumer, and this budget undercounts it. The check counts 1 + 8 bytes per variable-length key row plus the value length, and 40 bytes per range. It does not count:

the capacity-sized accumulator allocation itself;

the byte child, which starts at capacity * 4 and doubles on growth;

the child's nulls array, which is as large as its data;

the next accumulator, which is allocated eagerly after each push.

For a ~10-byte string key and capacity 4096, one accumulator actually holds ~168KB while ~78KB is counted (2.16x). At the 64MB default that is ~145MB per task, e.g. ~4.6GB on a 32-core executor. None of it is visible to the MemoryManager, and with off-heap column vectors none of it counts against spark.memory.offHeap.size either, so it shows up as a container kill or an OOM that cannot spill. The conf doc says "not the backing arrays", but not the 2-4x magnitude. Could this go through a MemoryConsumer? abandonSplicing() already maps naturally to spill(). Alternatively, the budget could at least count the real vector capacities.

dongjoon-hyun · 2026-09-25T21:15:22Z

+    // If needed, compute row indexes within a file. The row-index column is identified by name
+    // (ROW_INDEX_TEMPORARY_COLUMN_NAME), a synthetic metadata column no storage filter references,
+    // so its slot is always a non-key one and its ParquetColumnVector is always the persistent one.
+    if (rowIndexGenerator != null) {


This holds only for the synthetic metadata column. ParquetRowIndexUtil finds the row-index slot purely by name (_tmp_metadata_row_index), and that name is not reserved for data columns; ParquetFileMetadataStructRowIndexSuite even reads such a column. A real BIGINT data column with that name is in readDataColumns, so it can become a storage-filter key.

Its output then depends on how each row group was read:

Spliced row groups emit the file values from the survivor queues.

Row groups that gave up the filter or splicing emit the persistent vector, which populateRowIndex has overwritten with row indexes.

If the column is declared but missing from the file, the reader materializes row indexes while the filter is evaluated on null (the XxHash64 seed), so evalAllMissing can skip the whole file.

This is already broken without the feature (SPARK-40059), but now the result also varies with the conf and between row groups of one file. Excluding the row-index column from key slots, or rejecting it in storageFiltersFor, would keep it consistent.

dongjoon-hyun · 2026-09-25T21:15:22Z

+      // Closed at the end of the phase that reads it. `readFilteredRowGroup` hands out a store
+      // the file reader does not track, unlike `readNextRowGroup`, so nothing else would.
+      RowRanges survivors;
+      try (PageReadStore keyPages =


Minor, on spark.sql.files.ignoreCorruptFiles semantics: phase 1 decompresses and decodes every key page of a row group before any row of it is emitted. Page decompression is lazy in both paths, so the raw chunk read succeeds even when a page is corrupt. If a key page near the end of a row group is corrupt, the plain reader has already returned the batches before it, but this path returns none of that row group's rows. That differs from the documented "the contents that have been read will still be returned". The result shrinks rather than being wrong, so this may be acceptable, but it may deserve a note.

dongjoon-hyun · 2026-09-25T21:15:22Z

+        missing[i] = missingKeyLocalPositions.get(i);
+        missingValues[i] = existenceDefaults[keyIndices[missing[i]]];
+      }
+      storageFilter = storageFilter.rewriteForMissingKeys(missing, missingValues);


This re-derives what the reader materializes for a missing key column (the existence DEFAULT, else null) and rewrites the predicate with literals. That duplicates ParquetColumnVector's own missing-column rule. Because the rewrite rebuilds BloomFilterMightContain, it also needs the per-task rewrites cache and the "values are a function of the positions" invariant. I found no mismatch today, but if the two rules ever diverge, the filter would silently evaluate a value the scan never returns.

An alternative:

keep the predicate and keyColumnIndices unchanged;

put the constant/all-null persistent vectors that initBatch already built into the key-scratch batch's missing slots (reset() is a no-op for them);

skip missing keys when splicing.

evalAllMissing would then need to move after initBatch.

dongjoon-hyun · 2026-09-25T21:15:22Z

+        try {
+          dataPages = lateMatReader.readFilteredRowGroup(blockIdx, finalRanges);
+        } catch (MissingOffsetIndexException e) {
+          LOG.warn("Reading {} without page-level storage filtering: reading part of a row group "


A file without an offset index is an expected, documented degrade mode, yet this logs a WARN with a full stack trace (e) once per file, since fileHasNoOffsetIndex is per reader. On a table of thousands of pyarrow-written files, that floods the executor logs. parquet's own ColumnIndexFilter handles the same exception with a single INFO line and no stack trace.

Could this be INFO/DEBUG without the throwable, or logged once per executor? The same applies to logFilterGivenUpOnError, which is also suppressed only per file.

dongjoon-hyun · 2026-09-25T21:15:22Z

+      // projection is all keys and their values were buffered, since emit then builds every batch
+      // from the key queues alone.
+      long keptRows;
+      if (nonKeyColumns == null && spliceCurrentRowGroup) {


FileSourceStrategy.storageFiltersFor already returns Nil when every read data column is a key, so this all-keys mode (nonKeyColumns == null) is reachable only by driving the reader directly, as the comment at L806 says. Falling back with storageFilter = null; return; when nonKeyFieldCount == 0 would remove:

this branch;

the dataPages != null guard;

columns == null in compressedBytesForRowRanges.

Production behavior would not change. The catch is that the tests driving key-only reads (the key-type × dictionary matrix, etc.) would need a non-key column added. Without it, the tests that keep every row would pass silently without exercising the survivor queues.

dongjoon-hyun · 2026-09-25T21:15:22Z

+ * filtering did not save it. An all-keys projection has no page filtering at all, and
+ * [[rowsExcludedWithinRowGroup]] still counts every row the filter dropped.
+ */
+case class StorageFilterMetrics(


All five fields are nullable, and create has default arguments. However, the only production path (FileSourceScanLike.storageFilterMetrics → inputRDD → buildParquetReader) always supplies all five, as the reader's own comment at VectorizedParquetRecordReader.java:1002-1004 says. So several pieces exist only for tests:

the ~10 null checks in the reader;

the needBytes mode;

the five getOrElse(key, null) in ParquetFileFormat.

Making the fields required, and dropping the storageFilterMetrics = Map.empty default on FileFormat.buildReaderWithStorageFilters, would simplify this. Tests would then pass real SQLMetrics.

dongjoon-hyun · 2026-09-25T21:15:22Z

-  } ++ driverMetrics
+  } ++ storageFilterMetrics ++ driverMetrics
+
+  protected lazy val storageFilterMetrics: Map[String, SQLMetric] = if (storageFilters.nonEmpty) {


This defines Parquet-specific metric names ("row groups", "page filtering") in the format-neutral FileSourceScanLike, and ParquetFileFormat now imports the scan's companion to look them up with getOrElse(..., null). A second format that opts in (e.g. ORC) would show Parquet-named metrics it cannot fill.

The same principle this PR applies to the conf and to the expression checks (the format decides) suggests a FileFormat hook that returns the format's own metric map. That would keep the scan format-neutral and let the reader's metrics be non-optional. DSv2's supportedCustomMetrics / SQLMetrics.createV2CustomMetric is the precedent.

cloud-fan

Review summary

The previous review's blocking and non-blocking findings have been addressed in this revision. This pass found one non-blocking contract/correctness issue in the generic storage-filter handoff and four documentation or test-comment issues. No P0 or P1 findings remain. I did not run tests locally.

Findings

5 total: 0 P0, 0 P1, 1 P2, 4 P3.

Non-blocking (P2)

Preserve the accepted expression through reader construction — sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala:856 — see inline.

Nit (P3)

Rewrite the malformed configuration description — sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:1977 — see inline.
Describe the actual source of row indexes — sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java:28 — see inline.
Avoid attributing row IO to the exclusion metric — sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilter.scala:46 — see inline.
Align the suite comments with the tested contract — sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilterSuite.scala:349 — see inline.

Re-review status

Prior AI findings: 5 addressed, 0 still present; additional unresolved findings in this review: 5.

New attribution: 2 newly introduced, 3 late catch, 0 previously raised, 0 unattributed.

Remaining prior AI findings

No prior AI findings remain.

Existing discussions

existing discussion — This thread established that supportsStorageFilter and buildReaderWithStorageFilters must see the same original expression. The toSeq change fixes the first handoff, but preparedStorageFilters later binds attributes before the builder, which is a related new defect not covered by the original correction.

cloud-fan · 2026-09-26T02:40:50Z

+      val requestedDataAttrs = output.take(requiredSchema.length)
+      storageFilters.map { expr =>
+        val subqueryReplaced = expr.transform {  case s: execution.ScalarSubquery => s.toLiteral }
+        BindReferences.bindReference(subqueryReplaced, requestedDataAttrs)


Non-blocking (P2): supportsStorageFilter sees the original predicate, but this preparation replaces every AttributeReference with a BoundReference before buildReaderWithStorageFilters runs. That contradicts the documented same-expression handoff and means a format that opted in using a column name or field metadata cannot apply the same decision while constructing its reader. Could we carry both the original expression and an executable bound form (or otherwise preserve the advertised information)? Please also add an end-to-end generic FileFormat test that exercises the planner-to-reader handoff and the builder-returning-None fallback to buildReaderWithPartitionValues.

Recommended change: Represent the original offered expression and its executable prepared form explicitly at the FileFormat handoff, so the builder can observe the promised original information while Parquet still evaluates a materialized, bound form. Add generic FileFormat integration coverage for identity/metadata preservation and for None falling back to the ordinary reader.

Why this works: Prepare each accepted predicate without discarding its original representation, pass both representations through a single aligned handoff object or equivalent paired contract, and make Parquet consume the bound member. Exercise the full planning and scan-construction path with a test format that records the original expression and can decline specialized construction.

Scope: Refine the generic V1 storage-filter contract, its planner-to-scan transport, the Parquet consumer adaptation, and regression coverage.

Compatibility: Storage filtering remains a default-off optional optimization; unsupported formats and runtime give-up paths continue to return ordinary scan results.

Risks: The two representations could become misaligned if they are stored as independent sequences rather than one paired value. Changing the newly added FileFormat method shape must keep default decline behavior and Parquet serialization intact.

Constraints: Scalar subqueries must still be materialized before reader closure serialization. Parquet evaluation must remain bound to requiredSchema ordinals. The post-scan Filter remains authoritative and the optimization remains optional.

Success: A format can correlate the exact named/metadata-bearing expression it accepted with the executable predicate used to construct its reader. Parquet receives a materialized expression bound to the requested schema and preserves current results and fallback behavior. Returning None from the specialized builder invokes the ordinary builder and returns its rows. A regression that canonicalizes or discards the original expression before reader construction fails the new integration coverage.

cloud-fan · 2026-09-26T02:40:50Z

+  val PARQUET_STORAGE_FILTER_PUSHDOWN_MAX_SPLICED_ROW_GROUP_BYTES =
+    buildConf("spark.sql.parquet.storageFilterPushdown.maxSplicedRowGroupBytes")
+      .internal()
+      .doc("Most memory, in bytes, that the vectorized Parquet reader holds for one row group " +


Nit (P3): The opening Most memory ... is a sentence fragment, and the following independent clauses are joined with commas. Since this text appears in the generated configuration reference, could you rewrite it as a complete definition (for example, The maximum memory ...) and split the accounting rules into complete sentences?

cloud-fan · 2026-09-26T02:40:50Z

 final class ParquetReadState {
-  /** A special row range used when there is no row indexes (hence all rows must be included) */
-  private static final RowRange MAX_ROW_RANGE = new RowRange(Long.MIN_VALUE, Long.MAX_VALUE);
+  /** The row indexes to include, only not-null if the column index is present. */


Nit (P3): Non-null row indexes are not conditioned solely on a column index. The late-materialization path can pass finalRanges to readFilteredRowGroup, which derives row indexes from offset indexes even when the optional column index is absent. Could this comment describe the actual condition under which rowIndexes is populated?

cloud-fan · 2026-09-26T02:40:50Z

+ *    bytes phase 2 read, which is what `finalRanges` page selection pruned.
+ *
+ * The row counters' suffix says where a row was excluded, not what would have saved it: a row
+ * inside a kept row group is read as part of its page and dropped during decode, so page


Nit (P3): rowsExcludedWithinRowGroup counts all rows rejected inside a retained row group, but it does not prove that every rejected row's page was read: phase 2 can skip a page containing no surviving rows. Could this describe only what the counter establishes, without the row/page IO claim? The adjacent rowsExcludedByRowGroup bullet also needs a complete sentence.

cloud-fan · 2026-09-26T02:40:50Z

+
+  test("ParquetStorageFilter.create rejects a filter that violates a planner precondition") {
+    // These are all planner bugs by construction: storageFiltersFor pre-checks each one, and
+    // by the time create runs the conjunct is gone from the post-scan Filter, so a soft rejection


Nit (P3): Two comments in this suite describe a different contract from the one being tested: the suite exercises five storage-filter metrics rather than two, and FileSourceStrategy deliberately retains this conjunct in the post-scan Filter while also attaching it to the scan. Could you update both comments so they document the actual coverage and fallback-safety mechanism?

peter-toth force-pushed the SPARK-59620-storage-filter-pushdown branch from 7d8c03c to 8413639 Compare September 17, 2026 17:29

peter-toth changed the title ~~[SPARK-59620][SQL] Late-materialization storage-filter pushdown via splicing~~ [SPARK-59620][SQL] Late-materialization storage-filter pushdown in the vectorized Parquet reader Sep 17, 2026

peter-toth force-pushed the SPARK-59620-storage-filter-pushdown branch from 8413639 to 104d549 Compare September 22, 2026 12:05

peter-toth marked this pull request as ready for review September 22, 2026 12:05

dongjoon-hyun reviewed Sep 22, 2026

View reviewed changes

peter-toth marked this pull request as draft September 23, 2026 10:25

peter-toth marked this pull request as ready for review September 23, 2026 15:39

dongjoon-hyun reviewed Sep 23, 2026

View reviewed changes

peter-toth marked this pull request as draft September 23, 2026 17:01

peter-toth marked this pull request as ready for review September 24, 2026 18:42

dongjoon-hyun reviewed Sep 24, 2026

View reviewed changes

cloud-fan reviewed Sep 25, 2026

View reviewed changes

peter-toth added 3 commits September 25, 2026 17:57

Document that supportsStorageFilter sees the original expression

43c328c

dongjoon-hyun reviewed Sep 25, 2026

View reviewed changes

cloud-fan reviewed Sep 26, 2026

View reviewed changes

Conversation

peter-toth commented Sep 17, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Was this patch authored or co-authored using generative AI tooling?

Uh oh!

peter-toth commented Sep 17, 2026

Uh oh!

peter-toth commented Sep 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

peter-toth commented Sep 22, 2026

Uh oh!

dongjoon-hyun left a comment

Choose a reason for hiding this comment

Two resource-lifecycle bugs

requireOffsetIndexesForPhase2() makes this an availability feature, not a performance one

Unbounded, unaccounted survivor buffering

Design question: why buffer instead of re-read?

Smaller things

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

peter-toth commented Sep 23, 2026

Uh oh!

peter-toth commented Sep 23, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

peter-toth commented Sep 17, 2026 •

edited

Loading

peter-toth commented Sep 22, 2026 •

edited

Loading

`requireOffsetIndexesForPhase2()` makes this an availability feature, not a performance one